diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md
index 0aba2344..4dd19e87 100644
--- a/azure-functions-durable/CHANGELOG.md
+++ b/azure-functions-durable/CHANGELOG.md
@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ADDED
 
+- A best-effort background check warns when a Function App's requirements allow
+a future major version, which may introduce replay-breaking changes.
+
 - Added opt-in `HttpRetryOptions` on `context.call_http()` for durable initial
 request retries with configurable timing and status selection. Empty or omitted
 status lists select all non-2xx responses; exhausted retries fail the call.
diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md
index 03e90ec0..133d199a 100644
--- a/azure-functions-durable/README.md
+++ b/azure-functions-durable/README.md
@@ -17,9 +17,34 @@ built on top of the [`durabletask`](https://pypi.org/project/durabletask/) SDK.
 ## Installation
 
 ```bash
-pip install azure-functions-durable
+pip install "azure-functions-durable>=2.0.0b1,<3"
 ```
 
+For deployments, also constrain `azure-functions-durable` in the Function
+App's `requirements.txt`:
+
+```text
+azure-functions-durable>=2.0.0b1,<3
+```
+
+Future major versions may include broad, replay-breaking changes. An automatic
+upgrade during deployment can cause existing orchestrations to fail
+unexpectedly. Review the [migration guide](MIGRATION_GUIDE.md) and test replay
+before upgrading. A major-version constraint does not pin the `durabletask`
+dependency or guarantee replay compatibility; lock the complete dependency
+graph for reproducible deployments.
+
+After the first durable-client, orchestrator, or entity invocation,
+each Python worker checks accessible `requirements.txt` files in the
+background and writes a host worker log warning when the declared constraints
+permit a later major version. The check follows local `-r` and `-c` files and
+applicable environment markers. It is best-effort and may miss uncommon version
+ranges; missing or unsupported requirements are treated as unknown. A
+lockfile's presence does not indicate whether deployment uses it, so the warning
+describes the risk of installing directly from the requirements. The check does
+not read dependency files during app startup or wait for the check before
+returning from an invocation.
+
 ## Overview
 
 Author orchestrations, activities, and entities as Azure Functions and let the
diff --git a/azure-functions-durable/azure/durable_functions/internal/dependency_warning.py b/azure-functions-durable/azure/durable_functions/internal/dependency_warning.py
new file mode 100644
index 00000000..4d1b8d25
--- /dev/null
+++ b/azure-functions-durable/azure/durable_functions/internal/dependency_warning.py
@@ -0,0 +1,205 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Best-effort check for Function App dependencies that allow a new major."""
+
+import os
+import re
+import shlex
+from importlib.metadata import PackageNotFoundError, version
+from logging import Logger
+from pathlib import Path
+
+from packaging.markers import UndefinedComparison, UndefinedEnvironmentName
+from packaging.requirements import InvalidRequirement, Requirement
+from packaging.specifiers import SpecifierSet
+from packaging.utils import canonicalize_name
+from packaging.version import InvalidVersion, Version
+
+_PACKAGE = "azure-functions-durable"
+_INDEX_OPTIONS_WITH_VALUE = {
+    "-i", "-f", "--index-url", "--extra-index-url", "--find-links",
+    "--trusted-host", "--only-binary", "--no-binary",
+}
+_INDEX_FLAGS = {"--no-index", "--pre", "--prefer-binary"}
+
+
+def _read_specifiers(root: Path, logger: Logger) -> tuple[SpecifierSet, bool] | None:
+    specifiers: list[str] = []
+    required = False
+    allow_prereleases = False
+    active: set[tuple[Path, bool]] = set()
+    visited: set[tuple[Path, bool]] = set()
+
+    def read_file(path: Path, constraint: bool) -> bool:
+        nonlocal allow_prereleases, required
+        try:
+            path = path.resolve()
+            key = (path, constraint)
+            if key in active:
+                logger.debug("Cannot check Durable Functions dependencies: cyclic requirements in %s.", path)
+                return False
+            if key in visited:
+                return True
+            content = path.read_text(encoding="utf-8-sig")
+        except (OSError, UnicodeError) as exc:
+            logger.debug("Cannot read Function App requirements from %s: %s", path, exc)
+            return False
+
+        active.add(key)
+        try:
+            for raw_line in content.splitlines():
+                line = re.sub(r"\s+#.*$", "", raw_line).strip()
+                if not line or line.startswith("#"):
+                    continue
+                if line.endswith("\\"):
+                    logger.debug("Cannot check multiline requirements in %s.", path)
+                    return False
+                if line.startswith("-"):
+                    try:
+                        parts = shlex.split(line)
+                    except ValueError:
+                        logger.debug("Cannot parse a requirements option in %s.", path)
+                        return False
+                    option = parts[0]
+                    argument = None
+                    if option in ("-r", "--requirement", "-c", "--constraint") and len(parts) == 2:
+                        argument = parts[1]
+                    elif option.startswith(("--requirement=", "--constraint=")) and len(parts) == 1:
+                        option, argument = option.split("=", 1)
+                    elif option.startswith(("-r", "-c")) and len(option) > 2 and len(parts) == 1:
+                        option, argument = option[:2], option[2:]
+                    elif option in _INDEX_FLAGS and len(parts) == 1:
+                        allow_prereleases = allow_prereleases or option == "--pre"
+                        continue
+                    elif (
+                            option in _INDEX_OPTIONS_WITH_VALUE and len(parts) == 2) or (
+                            option.split("=", 1)[0] in _INDEX_OPTIONS_WITH_VALUE
+                            and "=" in option and len(parts) == 1):
+                        continue
+                    if not argument or "://" in argument or "${" in argument:
+                        logger.debug("Cannot inspect a requirements option in %s.", path)
+                        return False
+                    nested_constraint = option in ("-c", "--constraint")
+                    if not read_file(path.parent / argument, nested_constraint):
+                        return False
+                    continue
+                try:
+                    requirement = Requirement(line)
+                except InvalidRequirement:
+                    logger.debug("Cannot parse a requirement in %s.", path)
+                    return False
+                if canonicalize_name(requirement.name) != _PACKAGE:
+                    continue
+                try:
+                    if requirement.marker is not None and not requirement.marker.evaluate():
+                        continue
+                except (UndefinedComparison, UndefinedEnvironmentName, InvalidVersion) as exc:
+                    logger.debug("Cannot evaluate a requirement marker in %s: %s", path, exc)
+                    return False
+                if requirement.url is not None:
+                    logger.debug("Cannot determine the Durable Functions version from a URL in %s.", path)
+                    return False
+                required = required or not constraint
+                specifiers.append(str(requirement.specifier))
+        finally:
+            active.remove(key)
+        visited.add(key)
+        return True
+
+    if not read_file(root / "requirements.txt", False) or not required:
+        return None
+    package_specifiers = SpecifierSet(",".join(specifiers))
+    allow_prereleases = allow_prereleases or any(specifier.prereleases for specifier in package_specifiers)
+    return package_specifiers, allow_prereleases
+
+
+def _allows_later_major(
+        specifiers: SpecifierSet,
+        installed: Version,
+        allow_prereleases: bool = False) -> bool:
+    candidates: set[Version] = set()
+
+    def add_release(epoch: int, major: int, minor: int, patch: int) -> None:
+        prefix = f"{epoch}!" if epoch else ""
+        candidates.add(Version(f"{prefix}{major}.{minor}.{patch}"))
+
+    for major in (installed.major, installed.major + 1, installed.major + 2):
+        add_release(installed.epoch, major, 0, 0)
+        add_release(installed.epoch, major, 0, 1)
+        add_release(installed.epoch, major, 1, 0)
+
+    for specifier in specifiers:
+        try:
+            bound = Version(specifier.version.removesuffix(".*"))
+        except InvalidVersion:
+            continue
+        candidates.add(bound)
+        add_release(bound.epoch, bound.major, bound.minor, bound.micro + 1)
+        add_release(bound.epoch, bound.major, bound.minor + 1, 0)
+        add_release(bound.epoch, bound.major + 1, 0, 0)
+
+    for candidate in tuple(candidates):
+        base = candidate.base_version
+        candidates.update(Version(f"{base}{suffix}") for suffix in (".dev0", "a0", "b0", "rc0"))
+        if candidate.pre and candidate.pre[1] > 0:
+            label, number = candidate.pre
+            candidates.add(Version(f"{base}{label}{number - 1}"))
+
+    allowed = [
+        candidate for candidate in candidates
+        if specifiers.contains(candidate, prereleases=allow_prereleases)
+    ]
+    return (
+        any(candidate.major == installed.major for candidate in allowed)
+        and any(candidate.major > installed.major and candidate > installed for candidate in allowed)
+    )
+
+
+def warn_for_function_directory(function_directory: str, logger: Logger) -> None:
+    """Find the Function App root before checking its dependencies."""
+    roots: list[Path] = []
+    configured_root = os.environ.get("AzureWebJobsScriptRoot")
+    if configured_root:
+        roots.append(Path(configured_root))
+    directory = Path(function_directory)
+    roots.extend((directory, *directory.parents))
+    try:
+        roots.append(Path.cwd())
+    except OSError as exc:
+        logger.debug("Cannot determine the Function App working directory: %s", exc)
+    for root in roots:
+        if (root / "host.json").is_file():
+            warn_if_unbounded(str(root), logger)
+            return
+    logger.debug("Cannot find the Function App root for %s.", function_directory)
+
+
+def warn_if_unbounded(function_directory: str, logger: Logger) -> None:
+    """Warn when the app's requirements demonstrably permit a later major."""
+    root = Path(function_directory)
+    try:
+        requirements = _read_specifiers(root, logger)
+    except RecursionError as exc:
+        logger.debug("Cannot check deeply nested Function App requirements: %s", exc)
+        return
+    if requirements is None:
+        return
+    specifiers, allow_prereleases = requirements
+    try:
+        installed = Version(version(_PACKAGE))
+        allows_later_major = _allows_later_major(specifiers, installed, allow_prereleases)
+    except (PackageNotFoundError, InvalidVersion) as exc:
+        logger.debug("Cannot determine the installed Durable Functions version: %s", exc)
+        return
+    if allows_later_major:
+        next_major = f"{installed.epoch}!" if installed.epoch else ""
+        next_major += str(installed.major + 1)
+        logger.warning(
+            "The Function App requirements allow azure-functions-durable to upgrade beyond major version %d. "
+            "Future major releases may include broad, replay-breaking changes. If a deployment installs these "
+            "requirements without a lock, an automatic upgrade could cause existing orchestrations to fail "
+            "unexpectedly. Pin the major version, for example azure-functions-durable>=%s,<%s, "
+            "or deploy using a complete dependency lock.",
+            installed.major, installed.public, next_major,
+        )
diff --git a/azure-functions-durable/azure/durable_functions/internal/invocation.py b/azure-functions-durable/azure/durable_functions/internal/invocation.py
index 0d3eadea..47a50894 100644
--- a/azure-functions-durable/azure/durable_functions/internal/invocation.py
+++ b/azure-functions-durable/azure/durable_functions/internal/invocation.py
@@ -11,16 +11,50 @@
 from concurrent.futures import ThreadPoolExecutor
 from contextvars import ContextVar, copy_context
 from functools import lru_cache, wraps
+from threading import Lock, Thread
 from types import FunctionType
 from typing import Any, Callable, ParamSpec, TypeVar
 
 import azure.functions as func
 
 _invocation_context: ContextVar[func.Context | None] = ContextVar("durable_invocation_context", default=None)
+_dependency_check_lock = Lock()
+_dependency_check_scheduled = False
+_logger = logging.getLogger(__name__)
+# The worker routes SDK logs through the host even after an invocation ends.
+_dependency_logger = logging.getLogger("azure.functions.durable")
 _Parameters = ParamSpec("_Parameters")
 _Result = TypeVar("_Result")
 
 
+def _check_dependency_in_background(directory: str) -> None:
+    from .dependency_warning import warn_for_function_directory
+
+    warn_for_function_directory(directory, _dependency_logger)
+
+
+def _schedule_dependency_check(context: func.Context | None) -> None:
+    global _dependency_check_scheduled
+    if _dependency_check_scheduled or context is None:
+        return
+    directory = getattr(context, "function_directory", None)
+    if not isinstance(directory, str) or not directory:
+        return
+    with _dependency_check_lock:
+        if _dependency_check_scheduled:
+            return
+        _dependency_check_scheduled = True
+        try:
+            Thread(
+                target=_check_dependency_in_background,
+                args=(directory,),
+                name="durable-dependency-check",
+                daemon=True,
+            ).start()
+        except RuntimeError:
+            _logger.warning("Unable to start the Durable Functions dependency check.", exc_info=True)
+
+
 def preserve_function_source(wrapper: Callable[..., Any], original: Callable[..., Any]) -> None:
     """Preserve the filename used by the Functions worker to index the app directory."""
     if isinstance(wrapper, FunctionType) and isinstance(original, FunctionType):
@@ -131,6 +165,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                 return await function(*original.args, **original.kwargs)
             finally:
                 _invocation_context.reset(token)
+                _schedule_dependency_check(context)
 
         wrapper = async_wrapper
     else:
@@ -142,6 +177,7 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
                 return function(*original.args, **original.kwargs)
             finally:
                 _invocation_context.reset(token)
+                _schedule_dependency_check(context)
 
         wrapper = sync_wrapper
 
diff --git a/azure-functions-durable/pyproject.toml b/azure-functions-durable/pyproject.toml
index 7ea17a8f..7c3e5345 100644
--- a/azure-functions-durable/pyproject.toml
+++ b/azure-functions-durable/pyproject.toml
@@ -32,7 +32,8 @@ readme = "README.md"
 dependencies = [
     "durabletask[opentelemetry]>=1.13.1",
     "azure-identity>=1.19.0",
-    "azure-functions>=2.4.0b1"
+    "azure-functions>=2.4.0b1",
+    "packaging"
 ]
 
 [project.urls]
diff --git a/tests/azure-functions-durable/e2e/test_logging_e2e.py b/tests/azure-functions-durable/e2e/test_logging_e2e.py
index 84a71e4f..1eb62801 100644
--- a/tests/azure-functions-durable/e2e/test_logging_e2e.py
+++ b/tests/azure-functions-durable/e2e/test_logging_e2e.py
@@ -16,6 +16,14 @@ def _assert_host_routed(line: str, message: str) -> None:
     assert re.fullmatch(rf"\[[^\]]+Z\] {re.escape(message)}", line)
 
 
+def test_unpinned_dependency_warning_reaches_the_host(dtask_app):
+    dtask_app.start_orchestration("activity_chain")
+    warning = dtask_app.wait_for_host_log(
+        "The Function App requirements allow azure-functions-durable to upgrade beyond major version")
+    assert "replay-breaking changes" in warning
+    assert "existing orchestrations to fail unexpectedly" in warning
+
+
 def test_client_and_worker_logs_flow_through_host(dtask_app):
     instance_id = dtask_app.start_orchestration("activity_chain")
     status = dtask_app.wait_for_completion(instance_id)
diff --git a/tests/azure-functions-durable/test_dependency_warning.py b/tests/azure-functions-durable/test_dependency_warning.py
new file mode 100644
index 00000000..d542accc
--- /dev/null
+++ b/tests/azure-functions-durable/test_dependency_warning.py
@@ -0,0 +1,373 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Tests for deferred Function App dependency warnings."""
+
+import logging
+import subprocess
+import sys
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+
+import pytest
+
+from azure.durable_functions.internal import dependency_warning, invocation
+
+
+@pytest.fixture(autouse=True)
+def installed_version(monkeypatch):
+    monkeypatch.setattr(dependency_warning, "version", lambda name: "2.0.0rc5")
+
+
+def _warnings(caplog):
+    return [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING]
+
+
+@pytest.mark.parametrize(
+    ("requirement", "warns"),
+    [
+        ("azure-functions-durable", True),
+        ("azure-functions-durable>=2", True),
+        ("azure-functions-durable>=2.0.0b1", True),
+        ("azure-functions-durable>=2,<4", True),
+        ("azure-functions-durable>=2,<3.1", True),
+        ("azure-functions-durable>=2.0.0rc5,<3.0.0rc2", True),
+        ("azure-functions-durable>=2.0.0rc5,<3.0.0rc0", True),
+        ("azure-functions-durable>=2.0.0rc5,<3.0.0a0", True),
+        ("azure-functions-durable>=2,<=3,!=3", False),
+        ("--pre\nazure-functions-durable>=2.0.0rc5,<=3.0.0,!=3.0.0", True),
+        (
+            "azure-functions-durable>=2.0.0rc5,<3.0.0rc2,"
+            "!=3.0.0.dev0,!=3.0.0a0,!=3.0.0b0,!=3.0.0rc0",
+            True,
+        ),
+        ("azure-functions-durable!=3.*", True),
+        ("azure-functions-durable!=3.0.0", True),
+        ("azure-functions-durable<3", False),
+        ("azure-functions-durable>=2.0.0b1,<3", False),
+        ("azure-functions-durable>=2.0.0rc5,<3.0.0.dev0", False),
+        ("azure-functions-durable==2.*", False),
+        ("azure-functions-durable==2.0.0rc5", False),
+        ("azure-functions-durable~=2.0", False),
+        ("azure-functions-durable<=2.9.9", False),
+        ("azure-functions-durable<4,!=3.*", False),
+        ("azure-functions-durable>=3,<4", False),
+        ("Azure.Functions_Durable[extra]>=2.0.0b1,<3  # pinned", False),
+    ],
+)
+def test_requirements_warn_only_when_they_allow_a_later_major(tmp_path, caplog, requirement, warns):
+    (tmp_path / "requirements.txt").write_text(requirement, encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+
+    messages = _warnings(caplog)
+    assert bool(messages) is warns
+    if warns:
+        assert len(messages) == 1
+        assert "replay-breaking changes" in messages[0]
+        assert "an automatic upgrade could cause existing orchestrations to fail" in messages[0]
+        assert "azure-functions-durable>=2.0.0rc5,<3" in messages[0]
+
+
+def test_warning_tracks_the_installed_major(monkeypatch, tmp_path, caplog):
+    monkeypatch.setattr(dependency_warning, "version", lambda name: "3.2.1")
+    (tmp_path / "requirements.txt").write_text("azure-functions-durable>=3", encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert "azure-functions-durable>=3.2.1,<4" in _warnings(caplog)[0]
+
+
+def test_suggested_constraint_preserves_the_installed_epoch(monkeypatch, tmp_path, caplog):
+    monkeypatch.setattr(dependency_warning, "version", lambda name: "1!2.3")
+    (tmp_path / "requirements.txt").write_text("azure-functions-durable", encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert "azure-functions-durable>=1!2.3,<1!3" in _warnings(caplog)[0]
+
+
+def test_future_major_prerelease_preserves_the_installed_epoch(monkeypatch, tmp_path, caplog):
+    monkeypatch.setattr(dependency_warning, "version", lambda name: "1!2.0rc5")
+    (tmp_path / "requirements.txt").write_text(
+        "azure-functions-durable>=1!2.0rc5,<1!3.0rc2", encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert "azure-functions-durable>=1!2.0rc5,<1!3" in _warnings(caplog)[0]
+
+
+def test_a_higher_major_in_an_older_epoch_is_not_an_upgrade(monkeypatch, tmp_path, caplog):
+    monkeypatch.setattr(dependency_warning, "version", lambda name: "1!2.0")
+    (tmp_path / "requirements.txt").write_text("azure-functions-durable>=2,<1!3", encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert _warnings(caplog) == []
+
+
+@pytest.mark.parametrize(
+    ("contents", "warns"),
+    [
+        (
+            'azure-functions-durable>=2 ; python_version >= "3.0"\n'
+            'azure-functions-durable<3 ; python_version < "3.0"\n',
+            True,
+        ),
+        (
+            'azure-functions-durable>=2 ; python_version >= "3.0"\n'
+            'azure-functions-durable<3 ; python_version >= "3.0"\n',
+            False,
+        ),
+        (
+            'azure-functions-durable ; python_version < "3.0"\n',
+            False,
+        ),
+    ],
+)
+def test_environment_markers_apply_to_each_requirement(tmp_path, caplog, contents, warns):
+    (tmp_path / "requirements.txt").write_text(contents, encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert bool(_warnings(caplog)) is warns
+
+
+@pytest.mark.parametrize(
+    ("requirement_option", "constraint_option"),
+    [
+        ("-r nested/requirements.txt", "-c constraints.txt"),
+        ("--requirement=nested/requirements.txt", "--constraint=constraints.txt"),
+        ("-rnested/requirements.txt", "-cconstraints.txt"),
+    ],
+)
+def test_local_requirements_and_constraints_are_combined(
+        tmp_path, caplog, requirement_option, constraint_option):
+    (tmp_path / "requirements.txt").write_text(
+        f"{requirement_option}\n{constraint_option}\n", encoding="utf-8")
+    nested = tmp_path / "nested"
+    nested.mkdir()
+    (nested / "requirements.txt").write_text("azure-functions-durable\n", encoding="utf-8")
+    (tmp_path / "constraints.txt").write_text("azure-functions-durable<3\n", encoding="utf-8")
+
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert _warnings(caplog) == []
+
+    (tmp_path / "constraints.txt").write_text("azure-functions-durable>=2\n", encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert len(_warnings(caplog)) == 1
+
+
+def test_requirement_include_inside_constraint_resets_constraint_mode(monkeypatch, tmp_path, caplog):
+    monkeypatch.setattr(dependency_warning, "version", lambda name: "2.1.0")
+    (tmp_path / "requirements.txt").write_text("-c constraints.txt\n", encoding="utf-8")
+    (tmp_path / "constraints.txt").write_text("-r actual.txt\n", encoding="utf-8")
+    (tmp_path / "actual.txt").write_text("azure-functions-durable>=2.1.0\n", encoding="utf-8")
+
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+
+    assert len(_warnings(caplog)) == 1
+
+
+def test_pre_option_controls_candidate_prerelease_warnings(monkeypatch, tmp_path, caplog):
+    monkeypatch.setattr(dependency_warning, "version", lambda name: "2.1.0")
+    (tmp_path / "requirements.txt").write_text(
+        "azure-functions-durable>=2,<=3,!=3\n", encoding="utf-8")
+
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert _warnings(caplog) == []
+
+    (tmp_path / "requirements.txt").write_text(
+        "--pre\nazure-functions-durable>=2,<=3,!=3\n", encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert len(_warnings(caplog)) == 1
+
+
+def test_duplicate_unbounded_requirements_are_not_a_pin(tmp_path, caplog):
+    (tmp_path / "requirements.txt").write_text(
+        "azure-functions-durable\nazure-functions-durable\n", encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert len(_warnings(caplog)) == 1
+
+
+@pytest.mark.parametrize("option", ["--no-index", "--pre", "-i https://example.invalid", "--index-url=https://example.invalid"])
+def test_standalone_index_options_do_not_hide_unpinned_requirements(tmp_path, caplog, option):
+    (tmp_path / "requirements.txt").write_text(f"{option}\nazure-functions-durable\n", encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert len(_warnings(caplog)) == 1
+
+
+def test_combined_index_options_do_not_hide_constraints(tmp_path, caplog):
+    (tmp_path / "requirements.txt").write_text(
+        "azure-functions-durable\n--no-index -c constraints.txt\n", encoding="utf-8")
+    (tmp_path / "constraints.txt").write_text("azure-functions-durable<3\n", encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert _warnings(caplog) == []
+
+
+@pytest.mark.parametrize("lockfile", ["Pipfile.lock", "poetry.lock", "uv.lock", "pdm.lock", "pylock.toml"])
+def test_existing_lockfile_does_not_hide_unpinned_requirements(tmp_path, caplog, lockfile):
+    (tmp_path / "requirements.txt").write_text("azure-functions-durable", encoding="utf-8")
+    (tmp_path / lockfile).write_text("", encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert len(_warnings(caplog)) == 1
+    assert "If a deployment installs these requirements without a lock" in _warnings(caplog)[0]
+
+
+@pytest.mark.parametrize(
+    "contents",
+    [
+        "azure-functions\n",
+        "azure-functions-durable\n-r missing.txt\n",
+        "azure-functions-durable\n-c https://example.invalid/constraints.txt\n",
+        "azure-functions-durable\n-e .\n",
+        "azure-functions-durable @ https://example.invalid/package.whl\n",
+        "azure-functions-durable\n-r requirements.txt\n",
+        "azure-functions-durable \\\n  --hash=sha256:example\n",
+        "azure-functions-durable\ninvalid requirement @@\n",
+        "azure-functions-durable; python_version ~= 'not-a-version'\n",
+    ],
+)
+def test_unknown_requirements_do_not_claim_the_app_is_unpinned(tmp_path, caplog, contents):
+    (tmp_path / "requirements.txt").write_text(contents, encoding="utf-8")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logging.getLogger(__name__))
+    assert _warnings(caplog) == []
+
+
+def test_unreadable_or_missing_requirements_are_diagnostic_only(tmp_path, caplog):
+    caplog.set_level(logging.DEBUG)
+    logger = logging.getLogger(__name__)
+    dependency_warning.warn_if_unbounded(str(tmp_path), logger)
+    assert "Cannot read Function App requirements" in caplog.text
+    (tmp_path / "requirements.txt").write_bytes(b"\xff")
+    dependency_warning.warn_if_unbounded(str(tmp_path), logger)
+    assert _warnings(caplog) == []
+
+
+def test_app_root_prefers_the_function_directory_over_an_unrelated_working_directory(monkeypatch, tmp_path):
+    app = tmp_path / "app"
+    other = tmp_path / "other"
+    nested = app / "functions"
+    nested.mkdir(parents=True)
+    other.mkdir()
+    (app / "host.json").touch()
+    (other / "host.json").touch()
+    monkeypatch.chdir(other)
+    monkeypatch.delenv("AzureWebJobsScriptRoot", raising=False)
+    checked = []
+    monkeypatch.setattr(dependency_warning, "warn_if_unbounded", lambda root, logger: checked.append(root))
+
+    dependency_warning.warn_for_function_directory(str(nested), logging.getLogger(__name__))
+
+    assert checked == [str(app)]
+
+
+@pytest.mark.parametrize("configured", [False, True])
+def test_app_root_falls_back_to_the_configured_root_or_working_directory(monkeypatch, tmp_path, configured):
+    app = tmp_path / "app"
+    app.mkdir()
+    (app / "host.json").touch()
+    generated = tmp_path / "generated"
+    generated.mkdir()
+    monkeypatch.chdir(app if not configured else tmp_path)
+    if configured:
+        monkeypatch.setenv("AzureWebJobsScriptRoot", str(app))
+    else:
+        monkeypatch.delenv("AzureWebJobsScriptRoot", raising=False)
+    checked = []
+    monkeypatch.setattr(dependency_warning, "warn_if_unbounded", lambda root, logger: checked.append(root))
+
+    dependency_warning.warn_for_function_directory(str(generated), logging.getLogger(__name__))
+
+    assert checked == [str(app)]
+
+
+def test_background_check_uses_the_host_sdk_log_category(monkeypatch, tmp_path):
+    checked = []
+
+    def check(directory, logger):
+        checked.append((directory, logger.name))
+
+    monkeypatch.setattr(dependency_warning, "warn_for_function_directory", check)
+    invocation._check_dependency_in_background(str(tmp_path))
+    assert checked == [(str(tmp_path), "azure.functions.durable")]
+
+
+def test_app_indexing_does_not_import_the_dependency_check():
+    code = (
+        "import sys; import azure.durable_functions as df; df.DFApp(); "
+        "assert 'azure.durable_functions.internal.dependency_warning' not in sys.modules"
+    )
+    subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, check=True)
+
+
+def test_check_runs_once_in_the_background_after_an_invocation(monkeypatch, tmp_path):
+    started = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+    returned = threading.Event()
+    checked_directories = []
+    context = SimpleNamespace(
+        function_directory=str(tmp_path), invocation_id="invocation", thread_local_storage=threading.local())
+
+    def check(directory):
+        checked_directories.append(directory)
+        started.set()
+        try:
+            release.wait(timeout=10)
+        finally:
+            finished.set()
+
+    wrapped = invocation.wrap_invocation(lambda: "result")
+
+    def invoke():
+        assert wrapped(context=context) == "result"
+        returned.set()
+
+    monkeypatch.setattr(invocation, "_dependency_check_scheduled", False)
+    monkeypatch.setattr(invocation, "_check_dependency_in_background", check)
+    caller = threading.Thread(target=invoke)
+    caller.start()
+    try:
+        assert started.wait(timeout=5)
+        assert returned.wait(timeout=2), "The invocation waited for dependency inspection"
+        with ThreadPoolExecutor(max_workers=8) as executor:
+            list(executor.map(lambda _: invoke(), range(32)))
+        assert checked_directories == [str(tmp_path)]
+    finally:
+        release.set()
+        caller.join(timeout=5)
+        assert finished.wait(timeout=5)
+
+
+def test_thread_start_failure_is_reported_once_without_failing_the_invocation(monkeypatch, tmp_path, caplog):
+    class FailingThread:
+        def __init__(self, **kwargs):
+            pass
+
+        def start(self):
+            raise RuntimeError("thread unavailable")
+
+    monkeypatch.setattr(invocation, "_dependency_check_scheduled", False)
+    monkeypatch.setattr(invocation, "Thread", FailingThread)
+    context = SimpleNamespace(
+        function_directory=str(tmp_path), invocation_id="invocation", thread_local_storage=threading.local())
+    wrapped = invocation.wrap_invocation(lambda: "result")
+    for _ in range(2):
+        assert wrapped(context=context) == "result"
+    assert len(_warnings(caplog)) == 1
+    assert "Unable to start the Durable Functions dependency check" in _warnings(caplog)[0]
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_failed_invocations_still_schedule_the_check(monkeypatch, tmp_path, use_async):
+    context = SimpleNamespace(
+        function_directory=str(tmp_path), invocation_id="invocation", thread_local_storage=threading.local())
+    checked = []
+    monkeypatch.setattr(invocation, "_schedule_dependency_check", checked.append)
+
+    def fail():
+        raise ValueError("invocation failed")
+
+    async def async_fail():
+        fail()
+
+    wrapped = invocation.wrap_invocation(async_fail if use_async else fail)
+    with pytest.raises(ValueError, match="invocation failed"):
+        if use_async:
+            await wrapped(context=context)
+        else:
+            wrapped(context=context)
+
+    assert checked == [context]
+    assert invocation.get_invocation_id() is None