From 551d03d24ab52022e558573190f5770983786bc8 Mon Sep 17 00:00:00 2001
From: Ahmed Muhsin <ahmedmuhsin@microsoft.com>
Date: Fri, 26 Jun 2026 13:37:08 -0400
Subject: [PATCH 1/8] feat(durabletask): surface HITL respond-URL addressing to
 workflow executors

Let a workflow notify a human reviewer (e.g. email an approval link) from inside the
graph, without the caller threading the instanceId/requestId by hand.

- durabletask: the orchestrator injects host_context {instance_id, workflow_name,
  request_path_prefix} into each activity input; CapturingRunnerContext surfaces it as
  host_metadata. No new core API.
- azurefunctions: WorkflowHitlContext.from_context(ctx) builds the canonical
  respond/status URLs (returns None in-process so callers degrade gracefully).
  Re-exported through the agent_framework.azure lazy namespace.
- Nested sub-workflows: the address context (root instance + workflow name +
  accumulated {executor}~{ordinal}~ prefix) propagates down call_sub_orchestrator via a
  new SUBWORKFLOW_ADDRESS_KEY marker, so an executor at any depth builds a URL that
  targets the addressable top-level instance with a qualified request id. The per-child
  ordinal matches the read-side enumerate() index used by the status/respond endpoints.
  The marker is stripped from untrusted input alongside SUBWORKFLOW_INPUT_KEY
  (confused-deputy / info-leak guard).
- Samples 12 and 13 reworked into the retry-safe two-step notify pattern: the emitter
  generates an explicit request id and a downstream NotifyExecutor builds the URL and
  notifies, so failed upstream retries never produce a dead link.

Tests: unit coverage for the metadata round-trip, address/ordinal agreement (fan-out at
depth and nested prefix accumulation), marker stripping, and URL building; integration
tests assert the helper-built URL equals the server respondUrl and resumes the run, for
both the flat (12) and nested (13) samples.
---
 .../__init__.py                               |   2 +
 .../_hitl_context.py                          | 169 ++++++++++++++++++
 .../test_12_workflow_hitl.py                  |  65 +++++++
 .../test_13_workflow_subworkflow_hitl.py      |  49 +++++
 .../azurefunctions/tests/test_hitl_context.py | 165 +++++++++++++++++
 .../core/agent_framework/azure/__init__.py    |   1 +
 .../core/agent_framework/azure/__init__.pyi   |   3 +-
 .../_workflows/activity.py                    |   5 +
 .../_workflows/orchestrator.py                | 108 ++++++++++-
 .../_workflows/runner_context.py              |  20 +++
 .../_workflows/serialization.py               |  28 ++-
 .../tests/test_subworkflow_orchestration.py   | 123 ++++++++++++-
 .../tests/test_workflow_activity.py           |  36 ++++
 .../tests/test_workflow_serialization.py      |  15 ++
 .../12_workflow_hitl/function_app.py          | 110 +++++++++++-
 .../13_subworkflow_hitl/function_app.py       |  93 +++++++++-
 16 files changed, 961 insertions(+), 31 deletions(-)
 create mode 100644 python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
 create mode 100644 python/packages/azurefunctions/tests/test_hitl_context.py

diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/__init__.py b/python/packages/azurefunctions/agent_framework_azurefunctions/__init__.py
index 9b3d3db27a..2976a38858 100644
--- a/python/packages/azurefunctions/agent_framework_azurefunctions/__init__.py
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/__init__.py
@@ -3,6 +3,7 @@
 import importlib.metadata
 
 from ._app import AgentFunctionApp
+from ._hitl_context import WorkflowHitlContext
 
 try:
     __version__ = importlib.metadata.version(__name__)
@@ -11,5 +12,6 @@
 
 __all__ = [
     "AgentFunctionApp",
+    "WorkflowHitlContext",
     "__version__",
 ]
diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
new file mode 100644
index 0000000000..be48cea35f
--- /dev/null
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
@@ -0,0 +1,169 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Human-in-the-loop (HITL) addressing helper for workflow executors.
+
+When a MAF :class:`~agent_framework.Workflow` runs on the Azure Functions durable
+host, an executor can ask a human for input via ``ctx.request_info(...)``. To notify
+that human out-of-band (for example by emailing them an approval link), the executor
+needs the orchestration's ``instanceId`` and the request's ``requestId`` so it can
+build the ``/respond`` URL the reviewer will POST back to.
+
+:class:`WorkflowHitlContext` packages that addressing. It reads the orchestration
+metadata the durable host surfaces on the executor's runner context (see
+``CapturingRunnerContext.host_metadata``) and builds the canonical respond/status
+URLs that :class:`~agent_framework_azurefunctions.AgentFunctionApp` exposes -- so the
+executor never has to thread the instance id or base URL by hand.
+
+Typical use, from inside a notify executor reached by an edge from the executor that
+called ``request_info``::
+
+    hitl = WorkflowHitlContext.from_context(ctx)
+    if hitl is not None:  # None when not on the Azure Functions durable host
+        url = hitl.build_respond_url(request_id)
+        send_email(to=reviewer, body=f"Approve or reject here: {url}")
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Any, cast
+
+# App setting carrying the function app's host (e.g. ``myapp.azurewebsites.net``).
+# Azure Functions sets this automatically in the cloud; for local ``func start`` runs
+# add it to the ``Values`` map in ``local.settings.json`` (e.g. ``localhost:7071``).
+WEBSITE_HOSTNAME_ENV = "WEBSITE_HOSTNAME"
+
+# Default Azure Functions HTTP route prefix; the workflow routes registered by
+# AgentFunctionApp live under ``/api/workflow/...`` to match it.
+_API_ROUTE_PREFIX = "api"
+
+
+@dataclass(frozen=True)
+class WorkflowHitlContext:
+    """Builds Azure Functions HITL respond/status URLs from inside a workflow executor.
+
+    Obtain one with :meth:`from_context`. It exposes the addressable *root*
+    orchestration's ``instance_id`` and ``workflow_name`` and builds the URLs an
+    external reviewer uses to resume the workflow. When the executor runs inside a
+    nested sub-workflow, ``request_path_prefix`` carries the ``{executor}~{ordinal}~``
+    hops from the root down to this level, so :meth:`build_respond_url` qualifies a bare
+    request id back to the top-level instance automatically. The base URL is resolved
+    lazily (see :attr:`base_url`) from an explicit override or the ``WEBSITE_HOSTNAME``
+    app setting.
+    """
+
+    instance_id: str
+    workflow_name: str
+    base_url_override: str | None = None
+    request_path_prefix: str = ""
+
+    @classmethod
+    def from_context(
+        cls,
+        ctx: Any,
+        *,
+        base_url: str | None = None,
+    ) -> WorkflowHitlContext | None:
+        """Build a HITL context from a workflow executor's ``WorkflowContext``.
+
+        Reads the orchestration metadata the durable host attached to the executor's
+        runner context. Returns ``None`` when that metadata is absent -- i.e. the same
+        executor is running in-process rather than on the Azure Functions durable host
+        -- so callers can skip notification and degrade gracefully.
+
+        Args:
+            ctx: The ``WorkflowContext`` passed to the executor's handler.
+            base_url: Optional explicit base URL (scheme + host, e.g.
+                ``https://contoso.example.com``). Use this when the public URL differs
+                from ``WEBSITE_HOSTNAME`` -- for example behind a custom domain or API
+                Management gateway, where ``WEBSITE_HOSTNAME`` still reports the default
+                ``*.azurewebsites.net`` host. When omitted, the base URL is resolved
+                from ``WEBSITE_HOSTNAME`` on first use.
+
+        Returns:
+            A :class:`WorkflowHitlContext`, or ``None`` if not running on a durable host.
+        """
+        runner_context = getattr(ctx, "_runner_context", None)
+        raw_metadata = getattr(runner_context, "host_metadata", None)
+        if not isinstance(raw_metadata, dict):
+            return None
+        metadata = cast("dict[str, Any]", raw_metadata)
+
+        instance_id = metadata.get("instance_id")
+        workflow_name = metadata.get("workflow_name")
+        if not isinstance(instance_id, str) or not isinstance(workflow_name, str):
+            return None
+
+        # Present when the executor runs inside a nested sub-workflow; absent/empty at
+        # the top level. Defaults to "" so the request id is used unqualified.
+        raw_prefix = metadata.get("request_path_prefix")
+        request_path_prefix = raw_prefix if isinstance(raw_prefix, str) else ""
+
+        return cls(
+            instance_id=instance_id,
+            workflow_name=workflow_name,
+            base_url_override=base_url,
+            request_path_prefix=request_path_prefix,
+        )
+
+    @property
+    def base_url(self) -> str:
+        """The scheme + host the respond/status URLs are built on (no trailing slash).
+
+        Resolution order: the explicit ``base_url`` passed to :meth:`from_context`, then
+        the ``WEBSITE_HOSTNAME`` app setting (``http`` for localhost, otherwise
+        ``https``).
+
+        Raises:
+            RuntimeError: If neither an override nor ``WEBSITE_HOSTNAME`` is available.
+        """
+        if self.base_url_override:
+            return self.base_url_override.rstrip("/")
+
+        hostname = os.environ.get(WEBSITE_HOSTNAME_ENV)
+        if not hostname:
+            raise RuntimeError(
+                "Cannot build a HITL URL: no base URL is available. Set the "
+                f"'{WEBSITE_HOSTNAME_ENV}' app setting (present automatically on Azure "
+                "Functions; add it to the 'Values' map in local.settings.json for local "
+                "`func start` runs, e.g. 'localhost:7071'), or pass base_url=... to "
+                "WorkflowHitlContext.from_context()."
+            )
+
+        # An override may already include a scheme; WEBSITE_HOSTNAME is host-only, so
+        # infer one (http for local loopback, https otherwise).
+        if hostname.startswith(("http://", "https://")):
+            return hostname.rstrip("/")
+        scheme = "http" if hostname.startswith(("localhost", "127.0.0.1")) else "https"
+        return f"{scheme}://{hostname.rstrip('/')}"
+
+    def build_respond_url(self, request_id: str) -> str:
+        """Build the URL a reviewer POSTs their response to, resuming the workflow.
+
+        Mirrors the ``respondUrl`` AgentFunctionApp returns from its run/status
+        endpoints: ``{base}/api/workflow/{name}/respond/{instanceId}/{requestId}``,
+        always targeting the addressable top-level instance.
+
+        Args:
+            request_id: The pending request's id -- the id passed to (or generated by)
+                ``ctx.request_info``. Pass the **bare** id even from inside a nested
+                sub-workflow: any :attr:`request_path_prefix` is prepended for you to
+                qualify it (``{executor}~{ordinal}~{requestId}``) back to the root.
+
+        Returns:
+            The fully-qualified respond URL.
+        """
+        qualified_id = f"{self.request_path_prefix}{request_id}"
+        return (
+            f"{self.base_url}/{_API_ROUTE_PREFIX}/workflow/"
+            f"{self.workflow_name}/respond/{self.instance_id}/{qualified_id}"
+        )
+
+    def build_status_url(self) -> str:
+        """Build the workflow status URL for this orchestration instance.
+
+        Returns ``{base}/api/workflow/{name}/status/{instanceId}``, the same endpoint
+        AgentFunctionApp exposes for polling runtime status and pending HITL requests.
+        """
+        return f"{self.base_url}/{_API_ROUTE_PREFIX}/workflow/{self.workflow_name}/status/{self.instance_id}"
diff --git a/python/packages/azurefunctions/tests/integration_tests/test_12_workflow_hitl.py b/python/packages/azurefunctions/tests/integration_tests/test_12_workflow_hitl.py
index aec5e6615e..2778c2cb28 100644
--- a/python/packages/azurefunctions/tests/integration_tests/test_12_workflow_hitl.py
+++ b/python/packages/azurefunctions/tests/integration_tests/test_12_workflow_hitl.py
@@ -20,9 +20,12 @@
 """
 
 import time
+import uuid
 
 import pytest
 
+from agent_framework_azurefunctions import WorkflowHitlContext
+
 # Module-level markers - applied to all tests in this file
 pytestmark = [
     pytest.mark.flaky,
@@ -214,6 +217,68 @@ def test_hitl_workflow_with_neutral_content(self) -> None:
         final_status = self.helper.wait_for_orchestration(data["statusQueryGetUri"])
         assert final_status["runtimeStatus"] == "Completed"
 
+    def test_hitl_notify_respond_url_matches_helper(self) -> None:
+        """The respond URL WorkflowHitlContext builds equals the one the server accepts.
+
+        This is the core guarantee of the in-workflow notify pattern: the URL an
+        executor builds (via ``WorkflowHitlContext`` -- the same one ``NotifyExecutor``
+        would email a reviewer) is byte-for-byte the canonical respond URL the status
+        endpoint exposes, and POSTing to it actually resumes the run.
+        """
+        payload = {
+            "content_id": "article-test-005",
+            "title": "Sustainable Gardening Basics",
+            "body": (
+                "Composting kitchen scraps enriches soil naturally and reduces waste. "
+                "Rotating crops each season helps prevent nutrient depletion."
+            ),
+            "author": "Green Thumb",
+        }
+
+        # Start orchestration
+        response = self.helper.post_json(f"{self.base_url}/api/workflow/{WORKFLOW_NAME}/run", payload)
+        assert response.status_code == 202
+        data = response.json()
+        instance_id = data["instanceId"]
+
+        # Wait for the workflow to reach the HITL pause point
+        status = self._wait_for_hitl_request(instance_id)
+        pending_requests = status.get("pendingHumanInputRequests", [])
+        assert len(pending_requests) > 0, "Expected pending HITL request"
+        pending = pending_requests[0]
+        request_id = pending["requestId"]
+
+        # The review executor now generates an explicit uuid4 and passes it to
+        # request_info, so the pending id round-trips as a valid UUID (not an opaque
+        # framework default).
+        uuid.UUID(request_id)  # raises ValueError if not a valid UUID
+
+        # The request originates in the executor that called request_info.
+        assert pending.get("sourceExecutor") == "human_review_executor"
+
+        # Build the respond URL the same way an in-workflow executor would, via the
+        # public helper, pointing it at this app's base URL. It must equal the
+        # server-exposed respondUrl exactly -- i.e. the link NotifyExecutor emails is
+        # the one the /respond endpoint honors.
+        hitl = WorkflowHitlContext(
+            instance_id=instance_id,
+            workflow_name=WORKFLOW_NAME,
+            base_url_override=self.base_url,
+        )
+        helper_url = hitl.build_respond_url(request_id)
+        assert helper_url == pending["respondUrl"]
+
+        # Responding via the helper-built URL resumes the workflow to completion.
+        approval_response = self.helper.post_json(
+            helper_url,
+            {"approved": True, "reviewer_notes": "Looks good."},
+        )
+        assert approval_response.status_code == 200
+
+        final_status = self.helper.wait_for_orchestration(data["statusQueryGetUri"])
+        assert final_status["runtimeStatus"] == "Completed"
+        assert "output" in final_status
+
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])
diff --git a/python/packages/azurefunctions/tests/integration_tests/test_13_workflow_subworkflow_hitl.py b/python/packages/azurefunctions/tests/integration_tests/test_13_workflow_subworkflow_hitl.py
index 1b94eba3b3..c7174fa462 100644
--- a/python/packages/azurefunctions/tests/integration_tests/test_13_workflow_subworkflow_hitl.py
+++ b/python/packages/azurefunctions/tests/integration_tests/test_13_workflow_subworkflow_hitl.py
@@ -22,9 +22,12 @@
 """
 
 import time
+import uuid
 
 import pytest
 
+from agent_framework_azurefunctions import WorkflowHitlContext
+
 # Module-level markers - applied to all tests in this file
 pytestmark = [
     pytest.mark.flaky,
@@ -145,6 +148,52 @@ def test_nested_hitl_rejection(self) -> None:
         assert final_status["runtimeStatus"] == "Completed"
         assert "REJECTED" in str(final_status.get("output")).upper()
 
+    def test_nested_notify_respond_url_matches_helper(self) -> None:
+        """The URL the nested NotifyExecutor builds equals the server's qualified respondUrl.
+
+        Proves the address-propagation path end-to-end: the inner ``notify`` executor
+        (running in the child orchestration) builds a respond URL from the propagated
+        root instance + ``review_sub~0~`` prefix, given only the inner bare request id,
+        and that URL is byte-for-byte the qualified ``respondUrl`` the top-level status
+        endpoint exposes -- and POSTing to it resumes the nested run.
+        """
+        data = self._start({
+            "content_id": "article-200",
+            "title": "Tide Pool Ecology",
+            "body": "Intertidal zones host a surprising diversity of resilient marine life.",
+        })
+        instance_id = data["instanceId"]
+
+        status = self._wait_for_hitl_request(instance_id)
+        pending = status.get("pendingHumanInputRequests", [])
+        assert len(pending) == 1
+        qualified_id = pending[0]["requestId"]
+
+        # The qualified id is ``review_sub~0~{bare}``; the inner review gate now
+        # generates an explicit uuid4 as the bare id.
+        prefix = f"{SUBWORKFLOW_NODE_ID}~0~"
+        assert qualified_id.startswith(prefix), qualified_id
+        bare_id = qualified_id[len(prefix) :]
+        uuid.UUID(bare_id)  # raises if the inner id is not a valid uuid4
+
+        # Rebuild the URL exactly as the nested NotifyExecutor does: the root instance
+        # and root workflow name, the propagated path prefix, and only the bare id.
+        hitl = WorkflowHitlContext(
+            instance_id=instance_id,
+            workflow_name=WORKFLOW_NAME,
+            base_url_override=self.base_url,
+            request_path_prefix=prefix,
+        )
+        helper_url = hitl.build_respond_url(bare_id)
+        assert helper_url == pending[0]["respondUrl"]
+
+        # Responding via the helper-built URL resumes the nested run to completion.
+        approve = self.helper.post_json(helper_url, {"approved": True, "reviewer_notes": "ok"})
+        assert approve.status_code == 200
+        final_status = self.helper.wait_for_orchestration(data["statusQueryGetUri"])
+        assert final_status["runtimeStatus"] == "Completed"
+        assert "APPROVED" in str(final_status.get("output")).upper()
+
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])
diff --git a/python/packages/azurefunctions/tests/test_hitl_context.py b/python/packages/azurefunctions/tests/test_hitl_context.py
new file mode 100644
index 0000000000..f6ee6e10d3
--- /dev/null
+++ b/python/packages/azurefunctions/tests/test_hitl_context.py
@@ -0,0 +1,165 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Unit tests for WorkflowHitlContext (HITL respond-URL helper)."""
+
+# pyright: reportPrivateUsage=false
+
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from agent_framework_azurefunctions import WorkflowHitlContext
+from agent_framework_azurefunctions._hitl_context import WEBSITE_HOSTNAME_ENV
+
+
+def _ctx(metadata: Any) -> SimpleNamespace:
+    """Build a stand-in WorkflowContext exposing ``_runner_context.host_metadata``."""
+    return SimpleNamespace(_runner_context=SimpleNamespace(host_metadata=metadata))
+
+
+class TestFromContext:
+    """Construction from a workflow executor's context."""
+
+    def test_returns_context_when_metadata_present(self) -> None:
+        hitl = WorkflowHitlContext.from_context(_ctx({"instance_id": "inst-1", "workflow_name": "content_moderation"}))
+        assert hitl is not None
+        assert hitl.instance_id == "inst-1"
+        assert hitl.workflow_name == "content_moderation"
+
+    def test_returns_none_when_no_runner_context(self) -> None:
+        # A bare object without _runner_context (e.g. an unexpected ctx) yields None.
+        assert WorkflowHitlContext.from_context(SimpleNamespace()) is None
+
+    def test_returns_none_when_metadata_absent(self) -> None:
+        # In-process RunnerContext has no host_metadata -> getattr default None.
+        assert WorkflowHitlContext.from_context(_ctx(None)) is None
+
+    def test_returns_none_when_metadata_not_a_dict(self) -> None:
+        assert WorkflowHitlContext.from_context(_ctx("not-a-dict")) is None
+
+    def test_returns_none_when_instance_id_missing(self) -> None:
+        assert WorkflowHitlContext.from_context(_ctx({"workflow_name": "wf"})) is None
+
+    def test_returns_none_when_workflow_name_missing(self) -> None:
+        assert WorkflowHitlContext.from_context(_ctx({"instance_id": "inst-1"})) is None
+
+
+class TestBaseUrl:
+    """base_url resolution from override and WEBSITE_HOSTNAME."""
+
+    def test_explicit_override_wins(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv(WEBSITE_HOSTNAME_ENV, "ignored.azurewebsites.net")
+        hitl = WorkflowHitlContext.from_context(
+            _ctx({"instance_id": "i", "workflow_name": "wf"}),
+            base_url="https://contoso.example.com/",
+        )
+        assert hitl is not None
+        # Trailing slash trimmed; override used verbatim over the env host.
+        assert hitl.base_url == "https://contoso.example.com"
+
+    def test_website_hostname_gets_https(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv(WEBSITE_HOSTNAME_ENV, "myapp.azurewebsites.net")
+        hitl = WorkflowHitlContext.from_context(_ctx({"instance_id": "i", "workflow_name": "wf"}))
+        assert hitl is not None
+        assert hitl.base_url == "https://myapp.azurewebsites.net"
+
+    def test_localhost_gets_http(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv(WEBSITE_HOSTNAME_ENV, "localhost:7071")
+        hitl = WorkflowHitlContext.from_context(_ctx({"instance_id": "i", "workflow_name": "wf"}))
+        assert hitl is not None
+        assert hitl.base_url == "http://localhost:7071"
+
+    def test_override_with_scheme_preserved(self) -> None:
+        hitl = WorkflowHitlContext.from_context(
+            _ctx({"instance_id": "i", "workflow_name": "wf"}),
+            base_url="http://127.0.0.1:7071",
+        )
+        assert hitl is not None
+        assert hitl.base_url == "http://127.0.0.1:7071"
+
+    def test_raises_when_no_base_url_available(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv(WEBSITE_HOSTNAME_ENV, raising=False)
+        hitl = WorkflowHitlContext.from_context(_ctx({"instance_id": "i", "workflow_name": "wf"}))
+        assert hitl is not None
+        with pytest.raises(RuntimeError, match=WEBSITE_HOSTNAME_ENV):
+            _ = hitl.base_url
+
+
+class TestUrlBuilders:
+    """respond/status URL shapes match the AgentFunctionApp routes."""
+
+    def test_build_respond_url(self) -> None:
+        hitl = WorkflowHitlContext.from_context(
+            _ctx({"instance_id": "inst-1", "workflow_name": "content_moderation"}),
+            base_url="https://app.example.com",
+        )
+        assert hitl is not None
+        assert hitl.build_respond_url("req-9") == (
+            "https://app.example.com/api/workflow/content_moderation/respond/inst-1/req-9"
+        )
+
+    def test_build_respond_url_accepts_qualified_id(self) -> None:
+        # A nested sub-workflow request id (executor~ordinal~rid) flows through unchanged.
+        hitl = WorkflowHitlContext.from_context(
+            _ctx({"instance_id": "inst-1", "workflow_name": "wf"}),
+            base_url="https://app.example.com",
+        )
+        assert hitl is not None
+        assert hitl.build_respond_url("reviewer~0~req-9") == (
+            "https://app.example.com/api/workflow/wf/respond/inst-1/reviewer~0~req-9"
+        )
+
+    def test_build_status_url(self) -> None:
+        hitl = WorkflowHitlContext.from_context(
+            _ctx({"instance_id": "inst-1", "workflow_name": "wf"}),
+            base_url="https://app.example.com",
+        )
+        assert hitl is not None
+        assert hitl.build_status_url() == "https://app.example.com/api/workflow/wf/status/inst-1"
+
+
+class TestNestedPrefix:
+    """request_path_prefix qualifies a bare request id back to the root instance."""
+
+    def test_prefix_read_from_metadata(self) -> None:
+        # host_metadata for a nested executor carries the root instance/workflow and the
+        # accumulated path prefix; instance_id/workflow_name are the *root* values.
+        hitl = WorkflowHitlContext.from_context(
+            _ctx({
+                "instance_id": "root-inst",
+                "workflow_name": "moderation_pipeline",
+                "request_path_prefix": "review_sub~0~",
+            }),
+            base_url="https://app.example.com",
+        )
+        assert hitl is not None
+        assert hitl.request_path_prefix == "review_sub~0~"
+        # A bare request id is qualified back to the top-level instance automatically.
+        assert hitl.build_respond_url("req-9") == (
+            "https://app.example.com/api/workflow/moderation_pipeline/respond/root-inst/review_sub~0~req-9"
+        )
+
+    def test_deep_prefix(self) -> None:
+        hitl = WorkflowHitlContext.from_context(
+            _ctx({
+                "instance_id": "root-inst",
+                "workflow_name": "wf",
+                "request_path_prefix": "outer~2~inner~1~",
+            }),
+            base_url="https://app.example.com",
+        )
+        assert hitl is not None
+        assert hitl.build_respond_url("rid") == (
+            "https://app.example.com/api/workflow/wf/respond/root-inst/outer~2~inner~1~rid"
+        )
+
+    def test_absent_prefix_defaults_empty(self) -> None:
+        # Top-level metadata may omit the key; the bare id is used unqualified.
+        hitl = WorkflowHitlContext.from_context(
+            _ctx({"instance_id": "inst-1", "workflow_name": "wf"}),
+            base_url="https://app.example.com",
+        )
+        assert hitl is not None
+        assert hitl.request_path_prefix == ""
+        assert hitl.build_respond_url("rid") == ("https://app.example.com/api/workflow/wf/respond/inst-1/rid")
diff --git a/python/packages/core/agent_framework/azure/__init__.py b/python/packages/core/agent_framework/azure/__init__.py
index ba6a0e996d..565bb51b9e 100644
--- a/python/packages/core/agent_framework/azure/__init__.py
+++ b/python/packages/core/agent_framework/azure/__init__.py
@@ -20,6 +20,7 @@
     "DurableAIAgentOrchestrationContext": ("agent_framework_durabletask", "agent-framework-durabletask"),
     "DurableAIAgentWorker": ("agent_framework_durabletask", "agent-framework-durabletask"),
     "DurableWorkflowClient": ("agent_framework_durabletask", "agent-framework-durabletask"),
+    "WorkflowHitlContext": ("agent_framework_azurefunctions", "agent-framework-azurefunctions"),
 }
 
 
diff --git a/python/packages/core/agent_framework/azure/__init__.pyi b/python/packages/core/agent_framework/azure/__init__.pyi
index 7a914cee51..1157875845 100644
--- a/python/packages/core/agent_framework/azure/__init__.pyi
+++ b/python/packages/core/agent_framework/azure/__init__.pyi
@@ -8,7 +8,7 @@ from agent_framework_azure_ai_search import (
     AzureAISearchSettings,
 )
 from agent_framework_azure_cosmos import CosmosHistoryProvider
-from agent_framework_azurefunctions import AgentFunctionApp
+from agent_framework_azurefunctions import AgentFunctionApp, WorkflowHitlContext
 from agent_framework_durabletask import (
     AgentCallbackContext,
     AgentResponseCallbackProtocol,
@@ -31,4 +31,5 @@ __all__ = [
     "DurableAIAgentOrchestrationContext",
     "DurableAIAgentWorker",
     "DurableWorkflowClient",
+    "WorkflowHitlContext",
 ]
diff --git a/python/packages/durabletask/agent_framework_durabletask/_workflows/activity.py b/python/packages/durabletask/agent_framework_durabletask/_workflows/activity.py
index fb031039e1..821d19201e 100644
--- a/python/packages/durabletask/agent_framework_durabletask/_workflows/activity.py
+++ b/python/packages/durabletask/agent_framework_durabletask/_workflows/activity.py
@@ -67,6 +67,10 @@ def execute_workflow_activity(executor: Executor, input_json: str, workflow: Wor
     # omitted, so coerce None to the appropriate empty default.
     shared_state_snapshot: dict[str, Any] = data.get("shared_state_snapshot") or {}
     source_executor_ids = cast(list[str], data.get("source_executor_ids") or [SOURCE_ORCHESTRATOR])
+    # Orchestration metadata the orchestrator attaches when dispatching this activity
+    # (instance_id, workflow_name). Absent for in-process / legacy inputs, so default
+    # to None and surface it on the runner context for executors that want it.
+    host_context = cast("dict[str, Any] | None", data.get("host_context"))
 
     # Reconstruct the message - deserialize_value restores the original typed
     # objects from the encoded data (with type markers).
@@ -89,6 +93,7 @@ def classify_yielded_output(executor_id: str) -> YieldOutputEventType | None:
     async def _run() -> dict[str, Any]:
         runner_context = CapturingRunnerContext()
         runner_context.set_yield_output_classifier(classify_yielded_output)
+        runner_context.set_host_metadata(host_context)
         shared_state = State()
 
         # Deserialize shared state values to reconstruct dataclasses / Pydantic models.
diff --git a/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py b/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py
index d8ada5b593..6b438da406 100644
--- a/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py
+++ b/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py
@@ -50,8 +50,14 @@
 from agent_framework._workflows._state import State
 
 from .context import WorkflowOrchestrationContext
-from .naming import workflow_executor_activity_name, workflow_orchestrator_name, workflow_scoped_executor_id
+from .naming import (
+    qualify_subworkflow_request_id,
+    workflow_executor_activity_name,
+    workflow_orchestrator_name,
+    workflow_scoped_executor_id,
+)
 from .serialization import (
+    SUBWORKFLOW_ADDRESS_KEY,
     SUBWORKFLOW_INPUT_KEY,
     SUBWORKFLOW_RESULT_KEY,
     deserialize_value,
@@ -261,6 +267,7 @@ def _prepare_activity_task(
     source_executor_id: str,
     shared_state_snapshot: dict[str, Any] | None,
     workflow_name: str,
+    address: dict[str, str],
 ) -> Any:
     """Prepare an activity task for execution via the context adapter.
 
@@ -273,6 +280,17 @@ def _prepare_activity_task(
         "message": serialize_value(message),
         "shared_state_snapshot": shared_state_snapshot,
         "source_executor_ids": [source_executor_id],
+        # host_context addresses the *root* (HTTP-routable) orchestration so an executor
+        # can build a HITL respond URL (see CapturingRunnerContext.host_metadata):
+        # instance_id / workflow_name name the top-level instance, and
+        # request_path_prefix is the accumulated ``{executor}~{ordinal}~`` hops from the
+        # root down to this workflow level. For a top-level workflow the prefix is empty,
+        # so this reduces to addressing the instance directly.
+        "host_context": {
+            "instance_id": address["root_instance_id"],
+            "workflow_name": address["root_workflow_name"],
+            "request_path_prefix": address["request_path_prefix"],
+        },
     }
     activity_input_json = json.dumps(activity_input)
     activity_name = workflow_executor_activity_name(workflow_name, executor_id)
@@ -284,16 +302,23 @@ def _prepare_subworkflow_task(
     executor: WorkflowExecutor,
     message: Any,
     child_instance_id: str,
+    child_address: dict[str, str],
 ) -> Any:
     """Prepare a child-orchestration task that runs a ``WorkflowExecutor``'s inner workflow.
 
     The inner workflow runs as its own durable orchestration (``dafx-{innerName}``),
     so its executors are independently durable/observable. The node's message is
     serialized and wrapped in a marker so the child orchestrator reconstructs the
-    original typed object (trusted internal input).
+    original typed object (trusted internal input). A sibling address marker carries
+    the root instance / workflow name and this child's request-path prefix, so an
+    executor inside the child can build a respond URL that targets the top-level
+    instance with a qualified request id.
     """
     inner_orchestration_name = workflow_orchestrator_name(executor.workflow.name)
-    child_input = {SUBWORKFLOW_INPUT_KEY: serialize_value(message)}
+    child_input = {
+        SUBWORKFLOW_INPUT_KEY: serialize_value(message),
+        SUBWORKFLOW_ADDRESS_KEY: child_address,
+    }
     return ctx.call_sub_orchestrator(inner_orchestration_name, child_input, instance_id=child_instance_id)
 
 
@@ -637,6 +662,46 @@ def _try_unwrap_subworkflow_input(raw_value: Any) -> tuple[bool, Any]:
     return False, None
 
 
+def _resolve_workflow_address(initial_message: Any, instance_id: str, workflow_name: str) -> dict[str, str]:
+    """Resolve this orchestration's HITL address context.
+
+    Returns ``{root_instance_id, root_workflow_name, request_path_prefix}`` -- the
+    values an executor needs to build a respond URL that targets the addressable
+    top-level instance with a (possibly qualified) request id:
+
+    * A **child** orchestration receives its address from the parent in the
+      :data:`SUBWORKFLOW_ADDRESS_KEY` marker (the root instance/workflow plus the
+      ``{executor}~{ordinal}~`` path prefix down to this level), since its own
+      ``ctx.instance_id`` is a non-addressable child id.
+    * A **top-level** workflow has no such marker (it is stripped from untrusted input
+      at the host boundary by :func:`strip_subworkflow_markers`), so it is its own root
+      with an empty prefix.
+    """
+    if isinstance(initial_message, dict):
+        marker = cast("dict[str, Any]", initial_message)
+        addr = marker.get(SUBWORKFLOW_ADDRESS_KEY)
+        if isinstance(addr, dict):
+            typed = cast("dict[str, Any]", addr)
+            root_instance_id = typed.get("root_instance_id")
+            root_workflow_name = typed.get("root_workflow_name")
+            request_path_prefix = typed.get("request_path_prefix")
+            if (
+                isinstance(root_instance_id, str)
+                and isinstance(root_workflow_name, str)
+                and isinstance(request_path_prefix, str)
+            ):
+                return {
+                    "root_instance_id": root_instance_id,
+                    "root_workflow_name": root_workflow_name,
+                    "request_path_prefix": request_path_prefix,
+                }
+    return {
+        "root_instance_id": instance_id,
+        "root_workflow_name": workflow_name,
+        "request_path_prefix": "",
+    }
+
+
 def _coerce_initial_input(workflow: Workflow, raw_value: Any) -> Any:
     """Coerce the client's initial workflow input to the start executor's type.
 
@@ -779,6 +844,7 @@ def _prepare_all_tasks(
     pending_messages: dict[str, list[tuple[Any, str]]],
     shared_state: dict[str, Any] | None,
     subworkflow_counter: list[int],
+    address: dict[str, str],
 ) -> tuple[list[Any], list[TaskMetadata], list[tuple[str, Any, str]]]:
     """Prepare all pending tasks for parallel execution.
 
@@ -797,6 +863,10 @@ def _prepare_all_tasks(
         shared_state: Optional dict for cross-executor state sharing.
         subworkflow_counter: A single-element mutable counter, persistent across
             supersteps, used to derive unique deterministic child instance ids.
+        address: This orchestration's HITL address context
+            (``{root_instance_id, root_workflow_name, request_path_prefix}``). Surfaced
+            to activity executors via ``host_context`` and extended by one
+            ``{executor}~{ordinal}~`` hop for each dispatched sub-workflow child.
     """
     all_tasks: list[Any] = []
     task_metadata_list: list[TaskMetadata] = []
@@ -804,6 +874,14 @@ def _prepare_all_tasks(
 
     agent_messages_by_executor: dict[str, list[tuple[str, Any, str]]] = defaultdict(list)
 
+    # Per-executor, per-superstep ordinal for sub-workflow dispatch. This must match the
+    # read side's enumerate() index into the custom-status ``subworkflows[executorId]``
+    # list (built in this same dispatch order), so a nested pending request resolves
+    # back to the right child. It is deliberately distinct from ``subworkflow_counter``
+    # (a global, cross-superstep counter that only guarantees child-instance-id
+    # uniqueness, not addressing position).
+    per_executor_sub_ordinal: dict[str, int] = defaultdict(int)
+
     for executor_id, messages_with_sources in pending_messages.items():
         executor = workflow.executors[executor_id]
         is_agent = isinstance(executor, AgentExecutor)
@@ -819,8 +897,19 @@ def _prepare_all_tasks(
                 # are stable across orchestration replay.
                 child_instance_id = f"{ctx.instance_id}::{executor_id}::{subworkflow_counter[0]}"
                 subworkflow_counter[0] += 1
+                # Extend this orchestration's request-path prefix by one hop
+                # (``{executor}~{ordinal}~``) so an executor inside the child builds a
+                # respond URL qualified all the way back to the root instance.
+                ordinal = per_executor_sub_ordinal[executor_id]
+                per_executor_sub_ordinal[executor_id] += 1
+                child_address = {
+                    "root_instance_id": address["root_instance_id"],
+                    "root_workflow_name": address["root_workflow_name"],
+                    "request_path_prefix": address["request_path_prefix"]
+                    + qualify_subworkflow_request_id(executor_id, ordinal, ""),
+                }
                 logger.debug("Preparing sub-workflow task: %s -> %s", executor_id, child_instance_id)
-                task = _prepare_subworkflow_task(ctx, executor, message, child_instance_id)
+                task = _prepare_subworkflow_task(ctx, executor, message, child_instance_id, child_address)
                 all_tasks.append(task)
                 task_metadata_list.append(
                     TaskMetadata(
@@ -834,7 +923,7 @@ def _prepare_all_tasks(
             else:
                 logger.debug("Preparing activity task: %s", executor_id)
                 task = _prepare_activity_task(
-                    ctx, executor_id, message, source_executor_id, shared_state, workflow.name
+                    ctx, executor_id, message, source_executor_id, shared_state, workflow.name, address
                 )
                 all_tasks.append(task)
                 task_metadata_list.append(
@@ -919,6 +1008,13 @@ def run_workflow_orchestrator(
     # external client output path is unchanged.
     is_subworkflow = isinstance(initial_message, dict) and SUBWORKFLOW_INPUT_KEY in initial_message
 
+    # Resolve the HITL address context once: a child orchestration inherits the root
+    # instance/workflow + path prefix from the parent's address marker; a top-level
+    # workflow is its own root with an empty prefix. Threaded into task dispatch so an
+    # executor at any depth can build a respond URL targeting the addressable top-level
+    # instance.
+    workflow_address = _resolve_workflow_address(initial_message, ctx.instance_id, workflow.name)
+
     # Monotonic, replay-stable counter for deriving child orchestration instance ids;
     # persists across supersteps so repeated sub-workflow invocations never collide.
     subworkflow_counter: list[int] = [0]
@@ -992,7 +1088,7 @@ def publish_live_status(
 
         # Phase 1: Prepare all tasks
         all_tasks, task_metadata_list, remaining_agent_messages = _prepare_all_tasks(
-            ctx, workflow, pending_messages, shared_state, subworkflow_counter
+            ctx, workflow, pending_messages, shared_state, subworkflow_counter, workflow_address
         )
 
         # Agents and sub-workflows bypass the per-executor activity, so synthesize their
diff --git a/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py b/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py
index 1851339bb3..eae2f117d7 100644
--- a/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py
+++ b/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py
@@ -41,6 +41,7 @@ def __init__(self) -> None:
         self._workflow_id: str | None = None
         self._streaming: bool = False
         self._yield_output_classifier: YieldOutputClassifier = lambda _executor_id: "output"
+        self._host_metadata: dict[str, Any] | None = None
 
     # -- Messaging ------------------------------------------------------------
 
@@ -121,6 +122,25 @@ def set_streaming(self, streaming: bool) -> None:
     def is_streaming(self) -> bool:
         return self._streaming
 
+    # -- Host metadata --------------------------------------------------------
+
+    @property
+    def host_metadata(self) -> dict[str, Any] | None:
+        """Orchestration metadata injected by the durable host, or ``None`` in-process.
+
+        The durable orchestrator populates this from its own context (e.g. the
+        orchestration ``instance_id`` and ``workflow_name``) so an executor can
+        address the running orchestration -- for example to build a human-in-the-loop
+        respond URL and notify a reviewer -- without the caller threading those ids by
+        hand. It is ``None`` when the same executor runs in-process (no durable host),
+        so host-specific helpers can degrade gracefully.
+        """
+        return self._host_metadata
+
+    def set_host_metadata(self, metadata: dict[str, Any] | None) -> None:
+        """Set the orchestration metadata for this activity execution."""
+        self._host_metadata = metadata
+
     # -- Yield-output classification -------------------------------------------
 
     def set_yield_output_classifier(self, classifier: YieldOutputClassifier) -> None:
diff --git a/python/packages/durabletask/agent_framework_durabletask/_workflows/serialization.py b/python/packages/durabletask/agent_framework_durabletask/_workflows/serialization.py
index 680a614c6b..cabd2c4251 100644
--- a/python/packages/durabletask/agent_framework_durabletask/_workflows/serialization.py
+++ b/python/packages/durabletask/agent_framework_durabletask/_workflows/serialization.py
@@ -125,9 +125,18 @@ def strip_pickle_markers(data: Any) -> Any:
 # is unchanged. See ``orchestrator._process_subworkflow_result``.
 SUBWORKFLOW_RESULT_KEY = "__subworkflow_result__"
 
+# Alongside the input envelope, the parent passes the child its HITL *address* context
+# (the addressable root instance id, the root workflow name, and the accumulated
+# request-path prefix). This lets an executor inside a nested child build a respond URL
+# that targets the top-level instance with a qualified request id, without the child
+# knowing how its parent refers to it. Like SUBWORKFLOW_INPUT_KEY this is trusted
+# internal data (only call_sub_orchestrator sets it, post trust boundary), so it must
+# be stripped from untrusted top-level input (see strip_subworkflow_markers).
+SUBWORKFLOW_ADDRESS_KEY = "__subworkflow_address__"
+
 
 def strip_subworkflow_markers(data: Any) -> Any:
-    """Remove the reserved sub-workflow envelope key from untrusted top-level input.
+    """Remove the reserved sub-workflow envelope keys from untrusted top-level input.
 
     The orchestrator treats a top-level input dict carrying :data:`SUBWORKFLOW_INPUT_KEY`
     as a *trusted* child-orchestration payload and reconstructs it with
@@ -135,22 +144,25 @@ def strip_subworkflow_markers(data: Any) -> Any:
     :func:`strip_pickle_markers` sanitization, because a genuine envelope is only ever
     built internally (post trust boundary) by ``call_sub_orchestrator``. If untrusted
     client input could carry that key, an attacker could smuggle a pickle payload
-    straight into ``pickle.loads`` (RCE).
+    straight into ``pickle.loads`` (RCE). The sibling :data:`SUBWORKFLOW_ADDRESS_KEY`
+    is likewise trusted (it sets the root instance id / workflow name a nested executor
+    builds respond URLs against); if a top-level caller could supply it, they could
+    make the app emit URLs pointing at another instance (confused-deputy / info leak).
 
     Hosts therefore call this on client-supplied workflow input *before* scheduling the
-    orchestration, so the only way the orchestrator ever sees the envelope is from a
-    real internal child dispatch. Only the top-level key is removed (that is the only
-    position the orchestrator interprets it), leaving the rest of the caller's payload
-    untouched.
+    orchestration, so the only way the orchestrator ever sees either key is from a real
+    internal child dispatch. Only the top-level keys are removed (the only position the
+    orchestrator interprets them), leaving the rest of the caller's payload untouched.
     """
     if not isinstance(data, dict):
         return data
     typed = cast(dict[str, Any], data)
-    if SUBWORKFLOW_INPUT_KEY not in typed:
+    if SUBWORKFLOW_INPUT_KEY not in typed and SUBWORKFLOW_ADDRESS_KEY not in typed:
         return typed
-    logger.debug("Stripped reserved sub-workflow envelope key from untrusted input.")
+    logger.debug("Stripped reserved sub-workflow envelope key(s) from untrusted input.")
     cleaned = typed.copy()
     cleaned.pop(SUBWORKFLOW_INPUT_KEY, None)
+    cleaned.pop(SUBWORKFLOW_ADDRESS_KEY, None)
     return cleaned
 
 
diff --git a/python/packages/durabletask/tests/test_subworkflow_orchestration.py b/python/packages/durabletask/tests/test_subworkflow_orchestration.py
index 8ec735cdcd..850336ac1f 100644
--- a/python/packages/durabletask/tests/test_subworkflow_orchestration.py
+++ b/python/packages/durabletask/tests/test_subworkflow_orchestration.py
@@ -18,12 +18,16 @@
 
 from agent_framework import WorkflowExecutor
 
+from agent_framework_durabletask._workflows.naming import qualify_subworkflow_request_id
 from agent_framework_durabletask._workflows.orchestrator import (
+    SUBWORKFLOW_ADDRESS_KEY,
     SUBWORKFLOW_INPUT_KEY,
     TaskType,
     _coerce_initial_input,
+    _prepare_all_tasks,
     _prepare_subworkflow_task,
     _process_subworkflow_result,
+    _resolve_workflow_address,
     _try_unwrap_subworkflow_input,
     _unpack_subworkflow_result,
 )
@@ -57,6 +61,14 @@ def _result_envelope(outputs: list[Any], events: list[dict[str, Any]]) -> dict[s
     return {SUBWORKFLOW_RESULT_KEY: True, "outputs": outputs, "events": events}
 
 
+# A representative child address marker (root instance/workflow + one-hop path prefix).
+_CHILD_ADDRESS = {
+    "root_instance_id": "root-instance",
+    "root_workflow_name": "outer_wf",
+    "request_path_prefix": "sub-node~0~",
+}
+
+
 class TestPrepareSubworkflowTask:
     """Dispatch of a ``WorkflowExecutor`` node as a child orchestration."""
 
@@ -65,7 +77,7 @@ def test_schedules_inner_orchestration_by_scoped_name(self) -> None:
         ctx.call_sub_orchestrator.return_value = "task-sentinel"
         executor = _subworkflow_executor("sub-node", "inner_wf")
 
-        task = _prepare_subworkflow_task(ctx, executor, "hello", "parent::sub-node::0")
+        task = _prepare_subworkflow_task(ctx, executor, "hello", "parent::sub-node::0", _CHILD_ADDRESS)
 
         assert task == "task-sentinel"
         ctx.call_sub_orchestrator.assert_called_once()
@@ -77,12 +89,14 @@ def test_wraps_message_in_marker(self) -> None:
         ctx = Mock()
         executor = _subworkflow_executor("sub-node", "inner_wf")
 
-        _prepare_subworkflow_task(ctx, executor, "payload", "child-id")
+        _prepare_subworkflow_task(ctx, executor, "payload", "child-id", _CHILD_ADDRESS)
 
         args, _ = ctx.call_sub_orchestrator.call_args
         child_input = args[1]
         # The wrapped payload round-trips back to the original message.
         assert deserialize_value(child_input[SUBWORKFLOW_INPUT_KEY]) == "payload"
+        # The address marker rides alongside so the child can build respond URLs.
+        assert child_input[SUBWORKFLOW_ADDRESS_KEY] == _CHILD_ADDRESS
 
 
 class TestProcessSubworkflowResult:
@@ -251,3 +265,108 @@ def test_coerce_initial_input_returns_unwrapped_inner(self) -> None:
         marker = {SUBWORKFLOW_INPUT_KEY: "inner-message"}
 
         assert _coerce_initial_input(workflow, marker) == "inner-message"
+
+
+class TestResolveWorkflowAddress:
+    """Derivation of an orchestration's HITL address (root vs nested child)."""
+
+    def test_top_level_is_its_own_root_with_empty_prefix(self) -> None:
+        addr = _resolve_workflow_address("plain input", "top-instance", "outer_wf")
+        assert addr == {
+            "root_instance_id": "top-instance",
+            "root_workflow_name": "outer_wf",
+            "request_path_prefix": "",
+        }
+
+    def test_child_inherits_address_from_marker(self) -> None:
+        marker = {
+            SUBWORKFLOW_INPUT_KEY: "x",
+            SUBWORKFLOW_ADDRESS_KEY: {
+                "root_instance_id": "root",
+                "root_workflow_name": "outer_wf",
+                "request_path_prefix": "review_sub~0~",
+            },
+        }
+        # ctx.instance_id ("child-id") is ignored in favour of the marker's root values.
+        addr = _resolve_workflow_address(marker, "child-id", "human_review")
+        assert addr == {
+            "root_instance_id": "root",
+            "root_workflow_name": "outer_wf",
+            "request_path_prefix": "review_sub~0~",
+        }
+
+    def test_malformed_address_marker_falls_back_to_self(self) -> None:
+        # A non-dict / incomplete address marker is ignored (treated as top-level).
+        marker = {SUBWORKFLOW_ADDRESS_KEY: {"root_instance_id": 123}}
+        addr = _resolve_workflow_address(marker, "self-id", "wf")
+        assert addr == {"root_instance_id": "self-id", "root_workflow_name": "wf", "request_path_prefix": ""}
+
+
+class TestSubworkflowAddressPropagation:
+    """The dispatch-side request-path prefix must match the read-side qualified id.
+
+    The orchestrator builds each child's prefix from a *per-executor* ordinal; the
+    status/respond read path qualifies a nested request by ``enumerate()``-ing the
+    ``subworkflows[executorId]`` list. These two indexes must agree or an emailed
+    respond URL would resolve to the wrong child (or 404). This guards that agreement
+    for the tricky case: one node fanning out to several children in one superstep.
+    """
+
+    def _dispatch(self, address: dict[str, str], message_count: int) -> tuple[list[dict[str, str]], list[str]]:
+        """Run _prepare_all_tasks for one WorkflowExecutor node fanning out N children.
+
+        Returns (child_addresses, child_instance_ids) in dispatch order.
+        """
+        node_id = "review_sub"
+        executor = _subworkflow_executor(node_id, "human_review")
+        workflow = Mock()
+        workflow.executors = {node_id: executor}
+        workflow.name = "moderation_pipeline"
+
+        captured: list[dict[str, str]] = []
+
+        def _call_sub(name: str, input_: dict[str, object], *, instance_id: str) -> str:  # noqa: ARG001
+            captured.append(input_[SUBWORKFLOW_ADDRESS_KEY])  # type: ignore[arg-type]
+            return f"task::{instance_id}"
+
+        ctx = Mock()
+        ctx.instance_id = address["root_instance_id"]
+        ctx.call_sub_orchestrator.side_effect = _call_sub
+
+        pending = {node_id: [(f"msg-{i}", "src") for i in range(message_count)]}
+        _all_tasks, task_metadata, _remaining = _prepare_all_tasks(ctx, workflow, pending, None, [0], address)
+        child_ids = [m.child_instance_id for m in task_metadata if m.task_type == TaskType.SUBWORKFLOW]
+        assert all(cid is not None for cid in child_ids)
+        return captured, [cid for cid in child_ids if cid is not None]
+
+    def test_fanout_prefixes_match_readside_qualification(self) -> None:
+        top = {"root_instance_id": "root", "root_workflow_name": "moderation_pipeline", "request_path_prefix": ""}
+        addresses, child_ids = self._dispatch(top, message_count=3)
+
+        # Read side: subworkflows["review_sub"] = [child0, child1, child2]; a nested
+        # request from child ``ordinal`` is qualified as review_sub~{ordinal}~{bare}.
+        bare = "req-xyz"
+        for ordinal, child_address in enumerate(addresses):
+            dispatch_qualified = f"{child_address['request_path_prefix']}{bare}"
+            readside_qualified = qualify_subworkflow_request_id("review_sub", ordinal, bare)
+            assert dispatch_qualified == readside_qualified
+            # Root identity is carried through unchanged at every child.
+            assert child_address["root_instance_id"] == "root"
+            assert child_address["root_workflow_name"] == "moderation_pipeline"
+
+        # Child instance ids use the *global* counter (distinct from the ordinal).
+        assert child_ids == ["root::review_sub::0", "root::review_sub::1", "root::review_sub::2"]
+
+    def test_prefix_accumulates_when_already_nested(self) -> None:
+        # Simulate dispatching from a workflow that is itself one level deep.
+        nested = {
+            "root_instance_id": "root",
+            "root_workflow_name": "moderation_pipeline",
+            "request_path_prefix": "outer_node~2~",
+        }
+        addresses, _child_ids = self._dispatch(nested, message_count=2)
+
+        assert [a["request_path_prefix"] for a in addresses] == [
+            "outer_node~2~review_sub~0~",
+            "outer_node~2~review_sub~1~",
+        ]
diff --git a/python/packages/durabletask/tests/test_workflow_activity.py b/python/packages/durabletask/tests/test_workflow_activity.py
index 8271625021..de6650ff27 100644
--- a/python/packages/durabletask/tests/test_workflow_activity.py
+++ b/python/packages/durabletask/tests/test_workflow_activity.py
@@ -151,6 +151,42 @@ def test_hitl_response_handler_receives_typed_original_request() -> None:
     handler.assert_awaited_once()
 
 
+class TestExecuteWorkflowActivityHostMetadata:
+    """Orchestration metadata is surfaced to executors via the runner context."""
+
+    def test_host_context_surfaced_on_runner_context(self) -> None:
+        """``host_context`` in the activity input is exposed as ``runner_context.host_metadata``."""
+        captured: dict[str, Any] = {}
+
+        async def capture(message: Any, source_executor_ids: Any, state: Any, runner_context: Any) -> None:
+            captured["metadata"] = runner_context.host_metadata
+            state.commit()
+
+        executor = _make_executor("test-exec", capture)
+        input_data = json.dumps({
+            "message": "test",
+            "shared_state_snapshot": {},
+            "source_executor_ids": [SOURCE_ORCHESTRATOR],
+            "host_context": {"instance_id": "abc123", "workflow_name": "content_moderation"},
+        })
+        json.loads(execute_workflow_activity(executor, input_data))
+
+        assert captured["metadata"] == {"instance_id": "abc123", "workflow_name": "content_moderation"}
+
+    def test_absent_host_context_yields_none(self) -> None:
+        """When the input omits ``host_context``, ``host_metadata`` is ``None`` (in-process parity)."""
+        captured: dict[str, Any] = {}
+
+        async def capture(message: Any, source_executor_ids: Any, state: Any, runner_context: Any) -> None:
+            captured["metadata"] = runner_context.host_metadata
+            state.commit()
+
+        executor = _make_executor("test-exec", capture)
+        _run(executor, {})
+
+        assert captured["metadata"] is None
+
+
 if __name__ == "__main__":
     import pytest
 
diff --git a/python/packages/durabletask/tests/test_workflow_serialization.py b/python/packages/durabletask/tests/test_workflow_serialization.py
index 80295c9c80..381839b2c4 100644
--- a/python/packages/durabletask/tests/test_workflow_serialization.py
+++ b/python/packages/durabletask/tests/test_workflow_serialization.py
@@ -31,6 +31,7 @@
 from pydantic import BaseModel
 
 from agent_framework_durabletask._workflows.serialization import (
+    SUBWORKFLOW_ADDRESS_KEY,
     SUBWORKFLOW_INPUT_KEY,
     deserialize_value,
     deserialize_workflow_event,
@@ -437,6 +438,20 @@ def test_strips_input_key(self) -> None:
         data = {SUBWORKFLOW_INPUT_KEY: {"__pickled__": "evil"}, "real": 1}
         assert strip_subworkflow_markers(data) == {"real": 1}
 
+    def test_strips_address_key(self) -> None:
+        # A forged address marker would otherwise let a top-level caller point respond
+        # URLs at another instance (confused-deputy); it must be stripped too.
+        data = {SUBWORKFLOW_ADDRESS_KEY: {"root_instance_id": "victim"}, "real": 1}
+        assert strip_subworkflow_markers(data) == {"real": 1}
+
+    def test_strips_both_markers_together(self) -> None:
+        data = {
+            SUBWORKFLOW_INPUT_KEY: "x",
+            SUBWORKFLOW_ADDRESS_KEY: {"root_instance_id": "victim"},
+            "real": 1,
+        }
+        assert strip_subworkflow_markers(data) == {"real": 1}
+
     def test_strips_full_forged_envelope(self) -> None:
         data = {SUBWORKFLOW_INPUT_KEY: "x"}
         assert strip_subworkflow_markers(data) == {}
diff --git a/python/samples/04-hosting/azure_functions/12_workflow_hitl/function_app.py b/python/samples/04-hosting/azure_functions/12_workflow_hitl/function_app.py
index e51f2e98a8..499d66779b 100644
--- a/python/samples/04-hosting/azure_functions/12_workflow_hitl/function_app.py
+++ b/python/samples/04-hosting/azure_functions/12_workflow_hitl/function_app.py
@@ -8,11 +8,13 @@
 The workflow simulates a content moderation pipeline:
 1. User submits content for publication
 2. An AI agent analyzes the content for policy compliance
-3. A human reviewer is prompted to approve/reject the content
+3. A human reviewer is emailed an approval link and prompted to approve/reject
 4. Based on approval, content is either published or rejected
 
 Key architectural points:
 - Uses MAF's `ctx.request_info()` to pause workflow and request human input
+- A `NotifyExecutor` builds the respond URL via `WorkflowHitlContext` and notifies the
+  reviewer out-of-band (email) -- the caller never threads instanceId / requestId by hand
 - Uses `@response_handler` decorator to handle the human's response
 - AgentFunctionApp automatically provides HITL endpoints for status and response
 - Durable Functions provides durability while waiting for human input
@@ -27,6 +29,7 @@
 import os
 from dataclasses import dataclass
 from typing import Any
+from uuid import uuid4
 
 from agent_framework import (
     Agent,
@@ -42,7 +45,7 @@
 )
 from agent_framework.foundry import FoundryChatClient
 from agent_framework.openai import OpenAIChatOptions
-from agent_framework_azurefunctions import AgentFunctionApp
+from agent_framework_azurefunctions import AgentFunctionApp, WorkflowHitlContext
 from azure.identity.aio import AzureCliCredential
 from pydantic import BaseModel, ValidationError
 from typing_extensions import Never
@@ -117,6 +120,19 @@ class ModerationResult:
     reviewer_notes: str
 
 
+@dataclass
+class HumanReviewNotification:
+    """Payload handed to the notifier so it can build the respond URL and alert a human.
+
+    Carries the ``request_id`` the review executor passed to ``ctx.request_info`` so the
+    notifier addresses the exact pending request the workflow is waiting on.
+    """
+
+    request_id: str
+    content_id: str
+    prompt: str
+
+
 # ============================================================================
 # Agent Instructions
 # ============================================================================
@@ -196,14 +212,14 @@ def __init__(self):
     async def request_review(
         self,
         data: AnalysisWithSubmission,
-        ctx: WorkflowContext,
+        ctx: WorkflowContext[HumanReviewNotification],
     ) -> None:
         """Request human review for the content.
 
         This method:
         1. Constructs the approval request with all context
-        2. Calls request_info to pause the workflow
-        3. The workflow will resume when a response is provided via the HITL endpoint
+        2. Sends the request id to the notifier so it can email the reviewer a link
+        3. Calls request_info to pause the workflow until a response arrives
         """
         submission = data.submission
         analysis = data.analysis
@@ -234,11 +250,29 @@ async def request_review(
         # Store analysis in shared state for the response handler
         ctx.set_state("pending_analysis", data)
 
-        # Request human input - workflow will pause here
-        # The response_type specifies what we expect back
+        # Generate the request id up front so we can notify the reviewer *and* pause on
+        # the same id. The email is sent by a downstream executor (NotifyExecutor), not
+        # here, so only this activity's committed id ever reaches the notifier: if this
+        # executor is retried, the superseded attempts notify no one.
+        request_id = str(uuid4())
+
+        # Side-branch: hand the request id to the notifier so it can build the respond
+        # URL and alert the reviewer. This runs before the workflow pauses -- the
+        # orchestrator drains this message before entering the HITL wait.
+        await ctx.send_message(
+            HumanReviewNotification(
+                request_id=request_id,
+                content_id=submission.content_id,
+                prompt=prompt,
+            ),
+            target_id="notify_executor",
+        )
+
+        # Pause the workflow until a response arrives for this request id.
         await ctx.request_info(
             request_data=approval_request,
             response_type=HumanApprovalResponse,
+            request_id=request_id,
         )
 
     @response_handler
@@ -270,7 +304,61 @@ async def handle_approval_response(
             reviewer_notes=response.reviewer_notes,
         )
 
-        await ctx.send_message(result)
+        await ctx.send_message(result, target_id="publish_executor")
+
+
+class NotifyExecutor(Executor):
+    """Notifies a human reviewer with a respond link, without pausing the workflow.
+
+    Reached by an edge from ``HumanReviewExecutor`` in the same superstep that raises the
+    ``request_info`` request. It builds the canonical respond URL with
+    :class:`WorkflowHitlContext` and (in a real app) emails it to the reviewer. Because it
+    runs as a durable activity *downstream* of the id-generating executor, the id it emails
+    is always the one the orchestrator ends up waiting on -- retries of the upstream
+    executor never produce a dead link.
+    """
+
+    def __init__(self):
+        super().__init__(id="notify_executor")
+
+    @handler
+    async def notify(
+        self,
+        notification: HumanReviewNotification,
+        ctx: WorkflowContext,
+    ) -> None:
+        """Build the respond URL and notify the reviewer (logged here for the sample)."""
+        hitl = WorkflowHitlContext.from_context(ctx)
+        if hitl is None:
+            # Not running on the Azure Functions durable host (e.g. in-process DevUI):
+            # there is no respond endpoint to address, so skip notification.
+            logger.info(
+                "Not on the durable host; skipping reviewer notification for content %s.",
+                notification.content_id,
+            )
+            return
+
+        try:
+            respond_url = hitl.build_respond_url(notification.request_id)
+        except RuntimeError:
+            # No base URL configured (WEBSITE_HOSTNAME unset and no override). Notifying
+            # is a best-effort side effect -- the request is still reachable via the
+            # status endpoint -- so warn and continue rather than failing the workflow.
+            logger.warning(
+                "Cannot build a respond URL (set WEBSITE_HOSTNAME, e.g. in "
+                "local.settings.json). Skipping reviewer notification for content %s.",
+                notification.content_id,
+            )
+            return
+
+        # In a real application, send this URL to the reviewer (email, Teams, etc.). The
+        # reviewer POSTs {"approved": bool, "reviewer_notes": str} to it to resume the run.
+        logger.info(
+            "Human review needed for content %s.\n  Respond at: %s\n  Details: %s",
+            notification.content_id,
+            respond_url,
+            notification.prompt,
+        )
 
 
 class PublishExecutor(Executor):
@@ -381,17 +469,21 @@ def _create_workflow() -> Workflow:
     input_router = InputRouterExecutor()
     content_analyzer_executor = ContentAnalyzerExecutor()
     human_review_executor = HumanReviewExecutor()
+    notify_executor = NotifyExecutor()
     publish_executor = PublishExecutor()
 
     # Build the workflow graph
     # Flow:
     #   input_router -> content_analyzer_agent -> content_analyzer_executor
     #   -> human_review_executor (HITL pause here) -> publish_executor
+    # Side-branch: human_review_executor -> notify_executor emails the reviewer a respond
+    # link (built from WorkflowHitlContext) in the same superstep, before the pause.
     return (
         WorkflowBuilder(name="content_moderation", start_executor=input_router)
         .add_edge(input_router, content_analyzer_agent)
         .add_edge(content_analyzer_agent, content_analyzer_executor)
         .add_edge(content_analyzer_executor, human_review_executor)
+        .add_edge(human_review_executor, notify_executor)
         .add_edge(human_review_executor, publish_executor)
         .build()
     )
@@ -434,7 +526,7 @@ def launch(durable: bool = True) -> AgentFunctionApp | None:
     logger.info("- Human-in-the-loop using request_info / @response_handler pattern")
     logger.info("- AI content analysis with structured output")
     logger.info("- Human approval workflow integration")
-    logger.info("\nFlow: InputRouter -> ContentAnalyzer Agent -> HumanReview -> Publish")
+    logger.info("\nFlow: InputRouter -> ContentAnalyzer Agent -> HumanReview -> Notify/Publish")
 
     workflow = _create_workflow()
     serve(entities=[workflow], port=8096, auto_open=True)
diff --git a/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/function_app.py b/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/function_app.py
index bea2887f90..a35a88dafc 100644
--- a/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/function_app.py
+++ b/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/function_app.py
@@ -40,6 +40,7 @@
 
 import logging
 from dataclasses import dataclass
+from uuid import uuid4
 
 from agent_framework import (
     Executor,
@@ -50,7 +51,7 @@
     handler,
     response_handler,
 )
-from agent_framework_azurefunctions import AgentFunctionApp
+from agent_framework_azurefunctions import AgentFunctionApp, WorkflowHitlContext
 from pydantic import BaseModel
 from typing_extensions import Never
 
@@ -100,6 +101,19 @@ class ModerationDecision:
     reviewer_notes: str
 
 
+@dataclass
+class HumanReviewNotification:
+    """Payload handed to the notifier so it can build the (qualified) respond URL.
+
+    Carries the ``request_id`` the review gate passed to ``ctx.request_info`` so the
+    notifier addresses the exact pending request the (nested) workflow waits on.
+    """
+
+    request_id: str
+    content_id: str
+    prompt: str
+
+
 # ============================================================================
 # Inner workflow (contains the HITL pause)
 # ============================================================================
@@ -112,7 +126,9 @@ def __init__(self) -> None:
         super().__init__(id="review_gate")
 
     @handler
-    async def request_review(self, submission: ContentSubmission, ctx: WorkflowContext) -> None:
+    async def request_review(
+        self, submission: ContentSubmission, ctx: WorkflowContext[HumanReviewNotification]
+    ) -> None:
         prompt = (
             f"Please review the following content for publication:\n\n"
             f"Title: {submission.title}\n"
@@ -125,9 +141,29 @@ async def request_review(self, submission: ContentSubmission, ctx: WorkflowConte
             body=submission.body,
             prompt=prompt,
         )
+        # Generate the request id up front so we can notify the reviewer *and* pause on
+        # the same id. The notification is sent to a downstream executor, so only this
+        # activity's committed id ever reaches the notifier (retries notify no one).
+        request_id = str(uuid4())
+
+        # Side-branch: hand the request id to the notifier so it can build the respond
+        # URL. This runs before the (inner) workflow pauses.
+        await ctx.send_message(
+            HumanReviewNotification(
+                request_id=request_id,
+                content_id=submission.content_id,
+                prompt=prompt,
+            ),
+            target_id="notify",
+        )
+
         # Pause the (inner) workflow and wait for a human response. On the durable
         # host this pauses the child orchestration running this inner workflow.
-        await ctx.request_info(request_data=approval_request, response_type=HumanApprovalResponse)
+        await ctx.request_info(
+            request_data=approval_request,
+            response_type=HumanApprovalResponse,
+            request_id=request_id,
+        )
 
     @response_handler
     async def handle_approval_response(
@@ -152,10 +188,57 @@ async def handle_approval_response(
         )
 
 
+class NotifyExecutor(Executor):
+    """Inner-workflow notifier that builds the qualified respond URL for the reviewer.
+
+    Runs inside the nested ``human_review`` child orchestration, so the respond URL it
+    builds via :class:`WorkflowHitlContext` targets the **top-level** instance with a
+    qualified request id (``review_sub~0~{requestId}``) -- the address context is
+    propagated down from the parent, so this executor needs no knowledge of how it is
+    embedded.
+    """
+
+    def __init__(self) -> None:
+        super().__init__(id="notify")
+
+    @handler
+    async def notify(self, notification: HumanReviewNotification, ctx: WorkflowContext) -> None:
+        hitl = WorkflowHitlContext.from_context(ctx)
+        if hitl is None:
+            logger.info(
+                "Not on the durable host; skipping reviewer notification for content %s.",
+                notification.content_id,
+            )
+            return
+
+        try:
+            respond_url = hitl.build_respond_url(notification.request_id)
+        except RuntimeError:
+            # No base URL configured (WEBSITE_HOSTNAME unset and no override). Notifying
+            # is best-effort -- the request is still reachable via the status endpoint --
+            # so warn and continue rather than failing the workflow.
+            logger.warning(
+                "Cannot build a respond URL (set WEBSITE_HOSTNAME). Skipping reviewer notification for content %s.",
+                notification.content_id,
+            )
+            return
+
+        # In a real application, email/Teams this URL to the reviewer. They POST
+        # {"approved": bool, "reviewer_notes": str} to it to resume the run.
+        logger.info(
+            "Human review needed for content %s.\n  Respond at: %s",
+            notification.content_id,
+            respond_url,
+        )
+
+
 def create_inner_workflow() -> Workflow:
-    """Build the inner ``human_review`` workflow (a single HITL gate)."""
+    """Build the inner ``human_review`` workflow (HITL gate + reviewer notifier)."""
     review_gate = ReviewGateExecutor()
-    return WorkflowBuilder(name=INNER_WORKFLOW_NAME, start_executor=review_gate).build()
+    notify = NotifyExecutor()
+    # Side-branch: review_gate -> notify builds the qualified respond URL in the same
+    # superstep that raises the request, before the inner workflow pauses.
+    return WorkflowBuilder(name=INNER_WORKFLOW_NAME, start_executor=review_gate).add_edge(review_gate, notify).build()
 
 
 # ============================================================================

From ed7b3e1ef0645926aa172d1aabc4fab1918ade64 Mon Sep 17 00:00:00 2001
From: Ahmed Muhsin <ahmedmuhsin@microsoft.com>
Date: Fri, 26 Jun 2026 15:06:48 -0400
Subject: [PATCH 2/8] refactor(durabletask): read back request_info id instead
 of generating one in samples

Add WorkflowHitlContext.pending_request_id(ctx), an async helper that returns the
id request_info just generated (read from the runner context's pending request-info
events). This works on any host via the core RunnerContext protocol method, so it
needs no core change.

Samples 12 and 13 now call request_info() and read the id back to forward to the
NotifyExecutor, instead of minting a uuid by hand and passing request_id=. The
read-back happens in the same activity execution that generated the id, so the
pending request event and the notify message still commit together with the same id
(retry-safe; failed upstream retries notify no one).
---
 .../_hitl_context.py                          | 28 +++++++++++++++
 .../azurefunctions/tests/test_hitl_context.py | 36 +++++++++++++++++++
 .../12_workflow_hitl/function_app.py          | 28 +++++++--------
 .../13_subworkflow_hitl/function_app.py       | 27 +++++++-------
 4 files changed, 90 insertions(+), 29 deletions(-)

diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
index be48cea35f..f0daea2577 100644
--- a/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
@@ -107,6 +107,34 @@ def from_context(
             request_path_prefix=request_path_prefix,
         )
 
+    @staticmethod
+    async def pending_request_id(ctx: Any) -> str | None:
+        """Return the id of the most recently emitted ``request_info`` on ``ctx``.
+
+        Call this immediately after ``await ctx.request_info(...)`` to recover the
+        request id the framework generated, so it can be forwarded (e.g. in a message
+        to a downstream notify executor that builds the respond URL) without the caller
+        generating an id by hand.
+
+        It reads the executor's runner context, which both the in-process and durable
+        runners populate, so it works on any host and returns ``None`` only when no
+        request is pending (or the runner context does not track request-info events).
+
+        Note:
+            When an executor emits several ``request_info`` calls in one turn this
+            returns the **most recent** one. Read it right after each call -- or pass an
+            explicit ``request_id`` to ``request_info`` -- to disambiguate.
+        """
+        runner_context = getattr(ctx, "_runner_context", None)
+        getter = getattr(runner_context, "get_pending_request_info_events", None)
+        if getter is None:
+            return None
+        events = await getter()
+        if not events:
+            return None
+        # Dicts preserve insertion order, so the last key is the most recent request.
+        return next(reversed(events))
+
     @property
     def base_url(self) -> str:
         """The scheme + host the respond/status URLs are built on (no trailing slash).
diff --git a/python/packages/azurefunctions/tests/test_hitl_context.py b/python/packages/azurefunctions/tests/test_hitl_context.py
index f6ee6e10d3..2b88874357 100644
--- a/python/packages/azurefunctions/tests/test_hitl_context.py
+++ b/python/packages/azurefunctions/tests/test_hitl_context.py
@@ -163,3 +163,39 @@ def test_absent_prefix_defaults_empty(self) -> None:
         assert hitl is not None
         assert hitl.request_path_prefix == ""
         assert hitl.build_respond_url("rid") == ("https://app.example.com/api/workflow/wf/respond/inst-1/rid")
+
+
+def _ctx_with_pending(pending: dict[str, Any] | None, *, has_getter: bool = True) -> SimpleNamespace:
+    """Build a ctx whose runner context returns the given pending request-info events."""
+    if not has_getter:
+        return SimpleNamespace(_runner_context=SimpleNamespace())
+
+    async def _get() -> dict[str, Any]:
+        return pending or {}
+
+    return SimpleNamespace(_runner_context=SimpleNamespace(get_pending_request_info_events=_get))
+
+
+class TestPendingRequestId:
+    """Reading back the framework-generated request id after request_info."""
+
+    async def test_returns_latest_request_id(self) -> None:
+        # Dicts preserve insertion order; the most recently emitted request wins.
+        ctx = _ctx_with_pending({"r1": object(), "r2": object()})
+        assert await WorkflowHitlContext.pending_request_id(ctx) == "r2"
+
+    async def test_returns_single_request_id(self) -> None:
+        ctx = _ctx_with_pending({"only-one": object()})
+        assert await WorkflowHitlContext.pending_request_id(ctx) == "only-one"
+
+    async def test_returns_none_when_no_pending(self) -> None:
+        ctx = _ctx_with_pending({})
+        assert await WorkflowHitlContext.pending_request_id(ctx) is None
+
+    async def test_returns_none_when_no_runner_context(self) -> None:
+        assert await WorkflowHitlContext.pending_request_id(SimpleNamespace()) is None
+
+    async def test_returns_none_when_getter_absent(self) -> None:
+        # A runner context that doesn't track request-info events degrades to None.
+        ctx = _ctx_with_pending(None, has_getter=False)
+        assert await WorkflowHitlContext.pending_request_id(ctx) is None
diff --git a/python/samples/04-hosting/azure_functions/12_workflow_hitl/function_app.py b/python/samples/04-hosting/azure_functions/12_workflow_hitl/function_app.py
index 499d66779b..37ea8fdf1a 100644
--- a/python/samples/04-hosting/azure_functions/12_workflow_hitl/function_app.py
+++ b/python/samples/04-hosting/azure_functions/12_workflow_hitl/function_app.py
@@ -29,7 +29,6 @@
 import os
 from dataclasses import dataclass
 from typing import Any
-from uuid import uuid4
 
 from agent_framework import (
     Agent,
@@ -250,15 +249,21 @@ async def request_review(
         # Store analysis in shared state for the response handler
         ctx.set_state("pending_analysis", data)
 
-        # Generate the request id up front so we can notify the reviewer *and* pause on
-        # the same id. The email is sent by a downstream executor (NotifyExecutor), not
-        # here, so only this activity's committed id ever reaches the notifier: if this
-        # executor is retried, the superseded attempts notify no one.
-        request_id = str(uuid4())
+        # Pause the workflow for human input. request_info generates the request id;
+        # read it back so a downstream NotifyExecutor can build the respond URL -- no
+        # need to mint an id by hand.
+        await ctx.request_info(
+            request_data=approval_request,
+            response_type=HumanApprovalResponse,
+        )
+        request_id = await WorkflowHitlContext.pending_request_id(ctx)
+        assert request_id is not None  # always set immediately after request_info
 
         # Side-branch: hand the request id to the notifier so it can build the respond
-        # URL and alert the reviewer. This runs before the workflow pauses -- the
-        # orchestrator drains this message before entering the HITL wait.
+        # URL and alert the reviewer. The email is sent by the downstream NotifyExecutor
+        # (a separate activity), so only this activity's committed id ever reaches the
+        # notifier -- retries of this executor notify no one. The orchestrator drains
+        # this message before entering the HITL wait.
         await ctx.send_message(
             HumanReviewNotification(
                 request_id=request_id,
@@ -268,13 +273,6 @@ async def request_review(
             target_id="notify_executor",
         )
 
-        # Pause the workflow until a response arrives for this request id.
-        await ctx.request_info(
-            request_data=approval_request,
-            response_type=HumanApprovalResponse,
-            request_id=request_id,
-        )
-
     @response_handler
     async def handle_approval_response(
         self,
diff --git a/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/function_app.py b/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/function_app.py
index a35a88dafc..1b6e0f9aa4 100644
--- a/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/function_app.py
+++ b/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/function_app.py
@@ -40,7 +40,6 @@
 
 import logging
 from dataclasses import dataclass
-from uuid import uuid4
 
 from agent_framework import (
     Executor,
@@ -141,13 +140,21 @@ async def request_review(
             body=submission.body,
             prompt=prompt,
         )
-        # Generate the request id up front so we can notify the reviewer *and* pause on
-        # the same id. The notification is sent to a downstream executor, so only this
-        # activity's committed id ever reaches the notifier (retries notify no one).
-        request_id = str(uuid4())
+        # Pause the (inner) workflow and wait for a human response. On the durable host
+        # this pauses the child orchestration running this inner workflow. request_info
+        # generates the request id; read it back so the downstream notifier can build
+        # the (qualified) respond URL -- no need to mint an id by hand.
+        await ctx.request_info(
+            request_data=approval_request,
+            response_type=HumanApprovalResponse,
+        )
+        request_id = await WorkflowHitlContext.pending_request_id(ctx)
+        assert request_id is not None  # always set immediately after request_info
 
         # Side-branch: hand the request id to the notifier so it can build the respond
-        # URL. This runs before the (inner) workflow pauses.
+        # URL. The notifier is a separate (downstream) activity, so only this activity's
+        # committed id reaches it -- retries notify no one. This message is drained
+        # before the (inner) workflow pauses.
         await ctx.send_message(
             HumanReviewNotification(
                 request_id=request_id,
@@ -157,14 +164,6 @@ async def request_review(
             target_id="notify",
         )
 
-        # Pause the (inner) workflow and wait for a human response. On the durable
-        # host this pauses the child orchestration running this inner workflow.
-        await ctx.request_info(
-            request_data=approval_request,
-            response_type=HumanApprovalResponse,
-            request_id=request_id,
-        )
-
     @response_handler
     async def handle_approval_response(
         self,

From 18d53ed79e1f35d0b7a75459ef5ea6fc74400ae2 Mon Sep 17 00:00:00 2001
From: Ahmed Muhsin <ahmedmuhsin@microsoft.com>
Date: Fri, 26 Jun 2026 16:48:37 -0400
Subject: [PATCH 3/8] docs(azurefunctions): document request_info id read-back
 and notify safety

Tighten pending_request_id docstring to require calling it immediately after request_info, and explain why that is safe on the durable host (each executor runs in its own activity with its own runner context, so the pending set only holds this executor's requests and the newest is the one just emitted). Document the two-step notify pattern in the 12 and 13 sample READMEs, including the downstream-notifier retry safety and the nested address-prefix propagation.
---
 .../_hitl_context.py                          | 23 ++++++++++-------
 .../12_workflow_hitl/README.md                | 25 +++++++++++++++++++
 .../13_subworkflow_hitl/README.md             | 19 ++++++++++++++
 3 files changed, 58 insertions(+), 9 deletions(-)

diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
index f0daea2577..59a8939aab 100644
--- a/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
@@ -111,19 +111,24 @@ def from_context(
     async def pending_request_id(ctx: Any) -> str | None:
         """Return the id of the most recently emitted ``request_info`` on ``ctx``.
 
-        Call this immediately after ``await ctx.request_info(...)`` to recover the
+        Call this **immediately after** ``await ctx.request_info(...)`` to recover the
         request id the framework generated, so it can be forwarded (e.g. in a message
         to a downstream notify executor that builds the respond URL) without the caller
         generating an id by hand.
 
-        It reads the executor's runner context, which both the in-process and durable
-        runners populate, so it works on any host and returns ``None`` only when no
-        request is pending (or the runner context does not track request-info events).
-
-        Note:
-            When an executor emits several ``request_info`` calls in one turn this
-            returns the **most recent** one. Read it right after each call -- or pass an
-            explicit ``request_id`` to ``request_info`` -- to disambiguate.
+        Why "immediately after" is the rule, and why it is safe on the durable host:
+        the returned id is simply the newest entry in the executor's pending
+        request-info set, so reading right after a call always yields *that* call's id.
+        On the Azure Functions durable host every executor runs in its own activity with
+        its own runner context, so that set only ever holds this executor's own
+        requests (never another executor's), and the request you just emitted is always
+        the latest. If a single executor emits several ``request_info`` calls in one
+        turn, read this after **each** call (the only case where reading once at the end
+        would lose the earlier ids); or pass an explicit ``request_id`` to
+        ``request_info`` to address them directly.
+
+        Returns ``None`` only when no request is pending (or the runner context does not
+        track request-info events, e.g. in process off the durable host).
         """
         runner_context = getattr(ctx, "_runner_context", None)
         getter = getattr(runner_context, "get_pending_request_info_events", None)
diff --git a/python/samples/04-hosting/azure_functions/12_workflow_hitl/README.md b/python/samples/04-hosting/azure_functions/12_workflow_hitl/README.md
index 60e575ad82..30acd3f686 100644
--- a/python/samples/04-hosting/azure_functions/12_workflow_hitl/README.md
+++ b/python/samples/04-hosting/azure_functions/12_workflow_hitl/README.md
@@ -37,6 +37,31 @@ async def handle_approval_response(
     ...
 ```
 
+### Notifying a reviewer from inside the workflow
+
+This sample also shows how a workflow notifies a human itself (for example to email an approval link) instead of relying on the caller to poll the status endpoint. It uses a two step pattern so the reviewer gets a working respond URL.
+
+```python
+# Pause the workflow. request_info generates the request id internally.
+await ctx.request_info(
+    request_data=HumanApprovalRequest(...),
+    response_type=HumanApprovalResponse,
+)
+
+# Read that id back, then build the respond URL with WorkflowHitlContext.
+request_id = await WorkflowHitlContext.pending_request_id(ctx)
+hitl = WorkflowHitlContext.from_context(ctx)
+respond_url = hitl.build_respond_url(request_id)  # email this to the reviewer
+```
+
+Two things make this safe and worth understanding.
+
+**You never generate the id.** `request_info` already creates one and stores it on the context before it returns, so `pending_request_id(ctx)` reads it straight back. You must call `pending_request_id` immediately after `request_info`, because it returns the newest pending request, which is the one you just emitted. On the durable host this is reliable. Every executor runs in its own Durable Functions activity with its own runner context, so that pending set only ever holds this executor's own requests. If a single executor emits several requests in one turn, read the id after each call, or pass your own `request_id` to `request_info`.
+
+**The notification runs in a separate executor.** The review executor sends the id to a downstream `NotifyExecutor` that builds the URL, rather than emailing from the executor that generated the id. Because activities are at least once, an executor can be retried, and each retry mints a fresh id. Keeping the email in a downstream executor means only the committed attempt's id ever reaches the notifier, so a retried review executor never emails a dead link. `WorkflowHitlContext.from_context(ctx)` returns `None` when the executor is not on the durable host (for example under `--maf` DevUI), so the notify step skips cleanly there.
+
+The base URL comes from the `WEBSITE_HOSTNAME` app setting, which Azure Functions sets automatically in the cloud. For a custom domain or an API Management gateway, pass `base_url=...` to `from_context`, because `WEBSITE_HOSTNAME` still reports the default `*.azurewebsites.net` host.
+
 ### Automatic HITL Endpoints
 
 `AgentFunctionApp` automatically provides all the HTTP endpoints needed for HITL:
diff --git a/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/README.md b/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/README.md
index cdbbe9b368..e1bce74729 100644
--- a/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/README.md
+++ b/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/README.md
@@ -45,6 +45,25 @@ child orchestration, so a nested `request_info` is recorded on the *child* insta
 The separator is `~` (not `:`), so it never collides with framework-generated
 request ids such as functional-workflow `auto::N` ids.
 
+## Notifying a reviewer from inside a nested workflow
+
+This sample also notifies the reviewer from inside the inner workflow. The `review_gate` reads the id `request_info` generated and sends it to a downstream `NotifyExecutor`, which builds the respond URL with `WorkflowHitlContext`.
+
+```python
+# review_gate, inside the nested human_review workflow
+await ctx.request_info(request_data=HumanApprovalRequest(...), response_type=HumanApprovalResponse)
+request_id = await WorkflowHitlContext.pending_request_id(ctx)
+# ... send request_id to the notify executor ...
+
+# NotifyExecutor (also inside the inner workflow)
+hitl = WorkflowHitlContext.from_context(ctx)
+respond_url = hitl.build_respond_url(request_id)  # already qualified back to the root
+```
+
+The notify executor runs inside the child orchestration, yet `build_respond_url` returns a URL that targets the **top-level** instance with the qualified `review_sub~0~{requestId}` id. You pass only the **bare** inner id. The host propagates the address context (the root instance, the workflow name, and the `review_sub~0~` path prefix) down into the child, so the executor qualifies the id for you and never needs to know how it is embedded.
+
+The same two properties from the flat sample apply here. You never generate the id, because `request_info` creates it and `pending_request_id` reads it back, so call `pending_request_id` immediately after `request_info`. This is safe because each executor runs in its own Durable Functions activity with its own runner context, so the pending set only holds this executor's own requests and the newest one is the request you just emitted. And the email lives in a downstream executor, so a retried `review_gate` (activities are at least once) never emails a dead link, since only the committed attempt's id reaches the notifier.
+
 ## Endpoints
 
 `AgentFunctionApp` exposes routes only for the **top-level** workflow; the inner

From 4ddfd02d5382a10c2fea4cc9eaaec2ed5cd962e7 Mon Sep 17 00:00:00 2001
From: Ahmed Muhsin <ahmedmuhsin@microsoft.com>
Date: Wed, 8 Jul 2026 18:11:19 -0500
Subject: [PATCH 4/8] fix(python): resolve ty typing error and address PR
 review comments

- test_subworkflow_orchestration: replace the mypy-only type:ignore[arg-type] with a cast so the ty checker passes too (the other four checkers already honored the ignore).

- samples 12/13 README: guard the notify snippet against None before build_respond_url to match the documented graceful-degradation behavior.

- integration tests 12/13: reword comments that implied request_info now generates an explicit uuid4; it generates the id internally by default.
---
 .../tests/integration_tests/test_12_workflow_hitl.py        | 6 +++---
 .../integration_tests/test_13_workflow_subworkflow_hitl.py  | 4 ++--
 .../durabletask/tests/test_subworkflow_orchestration.py     | 4 ++--
 .../04-hosting/azure_functions/12_workflow_hitl/README.md   | 3 ++-
 .../azure_functions/13_subworkflow_hitl/README.md           | 3 ++-
 5 files changed, 11 insertions(+), 9 deletions(-)

diff --git a/python/packages/azurefunctions/tests/integration_tests/test_12_workflow_hitl.py b/python/packages/azurefunctions/tests/integration_tests/test_12_workflow_hitl.py
index 2778c2cb28..f018f07e86 100644
--- a/python/packages/azurefunctions/tests/integration_tests/test_12_workflow_hitl.py
+++ b/python/packages/azurefunctions/tests/integration_tests/test_12_workflow_hitl.py
@@ -248,9 +248,9 @@ def test_hitl_notify_respond_url_matches_helper(self) -> None:
         pending = pending_requests[0]
         request_id = pending["requestId"]
 
-        # The review executor now generates an explicit uuid4 and passes it to
-        # request_info, so the pending id round-trips as a valid UUID (not an opaque
-        # framework default).
+        # request_info generates the request id internally as a uuid4 when the caller
+        # does not pass one, so the pending id round-trips as a valid UUID (not an
+        # opaque framework default).
         uuid.UUID(request_id)  # raises ValueError if not a valid UUID
 
         # The request originates in the executor that called request_info.
diff --git a/python/packages/azurefunctions/tests/integration_tests/test_13_workflow_subworkflow_hitl.py b/python/packages/azurefunctions/tests/integration_tests/test_13_workflow_subworkflow_hitl.py
index c7174fa462..70d0a85675 100644
--- a/python/packages/azurefunctions/tests/integration_tests/test_13_workflow_subworkflow_hitl.py
+++ b/python/packages/azurefunctions/tests/integration_tests/test_13_workflow_subworkflow_hitl.py
@@ -169,8 +169,8 @@ def test_nested_notify_respond_url_matches_helper(self) -> None:
         assert len(pending) == 1
         qualified_id = pending[0]["requestId"]
 
-        # The qualified id is ``review_sub~0~{bare}``; the inner review gate now
-        # generates an explicit uuid4 as the bare id.
+        # The qualified id is ``review_sub~0~{bare}``; request_info generates the bare
+        # inner id internally as a uuid4 when the review gate does not pass one.
         prefix = f"{SUBWORKFLOW_NODE_ID}~0~"
         assert qualified_id.startswith(prefix), qualified_id
         bare_id = qualified_id[len(prefix) :]
diff --git a/python/packages/durabletask/tests/test_subworkflow_orchestration.py b/python/packages/durabletask/tests/test_subworkflow_orchestration.py
index 850336ac1f..588f5317db 100644
--- a/python/packages/durabletask/tests/test_subworkflow_orchestration.py
+++ b/python/packages/durabletask/tests/test_subworkflow_orchestration.py
@@ -13,7 +13,7 @@
   the original typed object on the child side.
 """
 
-from typing import Any
+from typing import Any, cast
 from unittest.mock import Mock
 
 from agent_framework import WorkflowExecutor
@@ -326,7 +326,7 @@ def _dispatch(self, address: dict[str, str], message_count: int) -> tuple[list[d
         captured: list[dict[str, str]] = []
 
         def _call_sub(name: str, input_: dict[str, object], *, instance_id: str) -> str:  # noqa: ARG001
-            captured.append(input_[SUBWORKFLOW_ADDRESS_KEY])  # type: ignore[arg-type]
+            captured.append(cast("dict[str, str]", input_[SUBWORKFLOW_ADDRESS_KEY]))
             return f"task::{instance_id}"
 
         ctx = Mock()
diff --git a/python/samples/04-hosting/azure_functions/12_workflow_hitl/README.md b/python/samples/04-hosting/azure_functions/12_workflow_hitl/README.md
index 30acd3f686..d7fa9ca735 100644
--- a/python/samples/04-hosting/azure_functions/12_workflow_hitl/README.md
+++ b/python/samples/04-hosting/azure_functions/12_workflow_hitl/README.md
@@ -51,7 +51,8 @@ await ctx.request_info(
 # Read that id back, then build the respond URL with WorkflowHitlContext.
 request_id = await WorkflowHitlContext.pending_request_id(ctx)
 hitl = WorkflowHitlContext.from_context(ctx)
-respond_url = hitl.build_respond_url(request_id)  # email this to the reviewer
+if hitl and request_id:
+    respond_url = hitl.build_respond_url(request_id)  # email this to the reviewer
 ```
 
 Two things make this safe and worth understanding.
diff --git a/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/README.md b/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/README.md
index e1bce74729..1cbfb7770d 100644
--- a/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/README.md
+++ b/python/samples/04-hosting/azure_functions/13_subworkflow_hitl/README.md
@@ -57,7 +57,8 @@ request_id = await WorkflowHitlContext.pending_request_id(ctx)
 
 # NotifyExecutor (also inside the inner workflow)
 hitl = WorkflowHitlContext.from_context(ctx)
-respond_url = hitl.build_respond_url(request_id)  # already qualified back to the root
+if hitl and request_id:
+    respond_url = hitl.build_respond_url(request_id)  # already qualified back to the root
 ```
 
 The notify executor runs inside the child orchestration, yet `build_respond_url` returns a URL that targets the **top-level** instance with the qualified `review_sub~0~{requestId}` id. You pass only the **bare** inner id. The host propagates the address context (the root instance, the workflow name, and the `review_sub~0~` path prefix) down into the child, so the executor qualifies the id for you and never needs to know how it is embedded.

From 1e208613266f986a81fa58834227e7466bc81fb6 Mon Sep 17 00:00:00 2001
From: Ahmed Muhsin <ahmedmuhsin@microsoft.com>
Date: Tue, 14 Jul 2026 18:35:44 -0500
Subject: [PATCH 5/8] fix(python): honor configurable Functions route prefix
 and address HITL PR review

- Resolve the route prefix from host.json (extensions.http.routePrefix, default api) in a new azurefunctions _routes module, used by both the server endpoints and WorkflowHitlContext, so a custom or empty routePrefix no longer 404s respond/status URLs (was hardcoded /api/ in four places).

- Extract respond/status URL construction into one shared builder called from _app.py and _hitl_context.py, removing the sync-by-test duplication.

- Broaden loopback detection (localhost, 127.0.0.0/8, 0.0.0.0, ::1, [::1]) via a _is_loopback helper so local links use http.

- Pin the host_context key names as shared constants in durabletask so producer and azurefunctions consumer cannot drift.

- Reword a stale base_url comment to reference WEBSITE_HOSTNAME.

- Add unit tests for the route module and loopback handling.
---
 .../agent_framework_azurefunctions/_app.py    |  16 +--
 .../_hitl_context.py                          |  57 +++++---
 .../agent_framework_azurefunctions/_routes.py | 112 ++++++++++++++++
 .../azurefunctions/tests/test_hitl_context.py |  32 ++++-
 .../azurefunctions/tests/test_routes.py       | 122 ++++++++++++++++++
 .../_workflows/orchestrator.py                |  11 +-
 .../_workflows/runner_context.py              |  10 ++
 7 files changed, 329 insertions(+), 31 deletions(-)
 create mode 100644 python/packages/azurefunctions/agent_framework_azurefunctions/_routes.py
 create mode 100644 python/packages/azurefunctions/tests/test_routes.py

diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py
index c8c2026eb6..868b11423b 100644
--- a/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py
@@ -57,6 +57,7 @@
 from ._entities import create_agent_entity
 from ._errors import IncomingRequestError
 from ._orchestration import AgentOrchestrationContextType, AgentTask, AzureFunctionsAgentExecutor
+from ._routes import build_workflow_respond_url, build_workflow_status_url, strip_route_prefix
 from ._workflow import run_workflow_orchestrator
 
 logger = logging.getLogger("agent_framework.azurefunctions")
@@ -507,13 +508,13 @@ async def start_workflow_orchestration(
             instance_id = await client.start_new(orchestrator_name, client_input=client_input)
 
             base_url = self._build_base_url(req.url)
-            status_url = f"{base_url}/api/workflow/{workflow_name}/status/{instance_id}"
+            status_url = build_workflow_status_url(base_url, workflow_name, instance_id)
 
             return func.HttpResponse(
                 json.dumps({
                     "instanceId": instance_id,
                     "statusQueryGetUri": status_url,
-                    "respondUri": f"{base_url}/api/workflow/{workflow_name}/respond/{instance_id}/{{requestId}}",
+                    "respondUri": build_workflow_respond_url(base_url, workflow_name, instance_id, "{requestId}"),
                     "message": "Workflow started",
                 }),
                 status_code=202,
@@ -574,8 +575,8 @@ async def get_workflow_status(
                             "requestData": req_data.get("data"),
                             "requestType": req_data.get("request_type"),
                             "responseType": req_data.get("response_type"),
-                            "respondUrl": (
-                                f"{base_url}/api/workflow/{workflow_name}/respond/{instance_id}/{qualified_id}"
+                            "respondUrl": build_workflow_respond_url(
+                                base_url, workflow_name, instance_id, qualified_id
                             ),
                         }
                         for qualified_id, req_data in gathered
@@ -741,11 +742,8 @@ async def _resolve_hitl_target(
         return await self._resolve_hitl_target(client, child_instance_id, remainder)
 
     def _build_base_url(self, request_url: str) -> str:
-        """Extract the base URL from a request URL."""
-        base_url, _, _ = request_url.partition("/api/")
-        if not base_url:
-            base_url = request_url.rstrip("/")
-        return base_url
+        """Extract the base URL (scheme + host) from a request URL, honoring the route prefix."""
+        return strip_route_prefix(request_url)
 
     def _is_owned_orchestration(self, status: Any, workflow_name: str) -> bool:
         """Return whether a durable orchestration status belongs to the named workflow.
diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
index 59a8939aab..ecc344ba02 100644
--- a/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
@@ -29,14 +29,36 @@
 from dataclasses import dataclass
 from typing import Any, cast
 
+from agent_framework_durabletask._workflows.runner_context import (
+    HOST_METADATA_INSTANCE_ID,
+    HOST_METADATA_REQUEST_PATH_PREFIX,
+    HOST_METADATA_WORKFLOW_NAME,
+)
+
+from ._routes import build_workflow_respond_url, build_workflow_status_url
+
 # App setting carrying the function app's host (e.g. ``myapp.azurewebsites.net``).
 # Azure Functions sets this automatically in the cloud; for local ``func start`` runs
 # add it to the ``Values`` map in ``local.settings.json`` (e.g. ``localhost:7071``).
 WEBSITE_HOSTNAME_ENV = "WEBSITE_HOSTNAME"
 
-# Default Azure Functions HTTP route prefix; the workflow routes registered by
-# AgentFunctionApp live under ``/api/workflow/...`` to match it.
-_API_ROUTE_PREFIX = "api"
+# Loopback hosts that resolve to ``http`` (not ``https``) when WEBSITE_HOSTNAME is
+# host-only; covers the addresses ``func start`` can bind locally.
+_LOOPBACK_HOSTS = frozenset({"localhost", "0.0.0.0", "::1"})  # noqa: S104
+
+
+def _is_loopback(host: str) -> bool:
+    """Return whether ``host`` (optionally ``host:port``) is a local loopback address.
+
+    Handles ``localhost``, IPv4 ``127.0.0.0/8`` and ``0.0.0.0``, and IPv6 ``::1``
+    (including the bracketed ``[::1]:port`` form ``func start`` prints).
+    """
+    normalized = host.strip().lower()
+    if normalized.startswith("["):  # bracketed IPv6 like [::1]:7071
+        normalized = normalized[1 : normalized.find("]")] if "]" in normalized else normalized[1:]
+    elif normalized.count(":") == 1:  # host:port (bare IPv6 has multiple colons)
+        normalized = normalized.split(":", 1)[0]
+    return normalized in _LOOPBACK_HOSTS or normalized.startswith("127.")
 
 
 @dataclass(frozen=True)
@@ -90,14 +112,14 @@ def from_context(
             return None
         metadata = cast("dict[str, Any]", raw_metadata)
 
-        instance_id = metadata.get("instance_id")
-        workflow_name = metadata.get("workflow_name")
+        instance_id = metadata.get(HOST_METADATA_INSTANCE_ID)
+        workflow_name = metadata.get(HOST_METADATA_WORKFLOW_NAME)
         if not isinstance(instance_id, str) or not isinstance(workflow_name, str):
             return None
 
         # Present when the executor runs inside a nested sub-workflow; absent/empty at
         # the top level. Defaults to "" so the request id is used unqualified.
-        raw_prefix = metadata.get("request_path_prefix")
+        raw_prefix = metadata.get(HOST_METADATA_REQUEST_PATH_PREFIX)
         request_path_prefix = raw_prefix if isinstance(raw_prefix, str) else ""
 
         return cls(
@@ -164,19 +186,20 @@ def base_url(self) -> str:
                 "WorkflowHitlContext.from_context()."
             )
 
-        # An override may already include a scheme; WEBSITE_HOSTNAME is host-only, so
-        # infer one (http for local loopback, https otherwise).
+        # WEBSITE_HOSTNAME may include a scheme (unusual but possible); otherwise it is
+        # host-only, so infer one (http for local loopback, https otherwise).
         if hostname.startswith(("http://", "https://")):
             return hostname.rstrip("/")
-        scheme = "http" if hostname.startswith(("localhost", "127.0.0.1")) else "https"
+        scheme = "http" if _is_loopback(hostname) else "https"
         return f"{scheme}://{hostname.rstrip('/')}"
 
     def build_respond_url(self, request_id: str) -> str:
         """Build the URL a reviewer POSTs their response to, resuming the workflow.
 
         Mirrors the ``respondUrl`` AgentFunctionApp returns from its run/status
-        endpoints: ``{base}/api/workflow/{name}/respond/{instanceId}/{requestId}``,
-        always targeting the addressable top-level instance.
+        endpoints: ``{base}/{prefix}/workflow/{name}/respond/{instanceId}/{requestId}``
+        (``prefix`` is the app's ``routePrefix``, ``api`` by default), always targeting
+        the addressable top-level instance.
 
         Args:
             request_id: The pending request's id -- the id passed to (or generated by)
@@ -188,15 +211,13 @@ def build_respond_url(self, request_id: str) -> str:
             The fully-qualified respond URL.
         """
         qualified_id = f"{self.request_path_prefix}{request_id}"
-        return (
-            f"{self.base_url}/{_API_ROUTE_PREFIX}/workflow/"
-            f"{self.workflow_name}/respond/{self.instance_id}/{qualified_id}"
-        )
+        return build_workflow_respond_url(self.base_url, self.workflow_name, self.instance_id, qualified_id)
 
     def build_status_url(self) -> str:
         """Build the workflow status URL for this orchestration instance.
 
-        Returns ``{base}/api/workflow/{name}/status/{instanceId}``, the same endpoint
-        AgentFunctionApp exposes for polling runtime status and pending HITL requests.
+        Returns ``{base}/{prefix}/workflow/{name}/status/{instanceId}`` (``prefix`` is the
+        app's ``routePrefix``, ``api`` by default), the same endpoint AgentFunctionApp
+        exposes for polling runtime status and pending HITL requests.
         """
-        return f"{self.base_url}/{_API_ROUTE_PREFIX}/workflow/{self.workflow_name}/status/{self.instance_id}"
+        return build_workflow_status_url(self.base_url, self.workflow_name, self.instance_id)
diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_routes.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_routes.py
new file mode 100644
index 0000000000..91c0a6394a
--- /dev/null
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_routes.py
@@ -0,0 +1,112 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Single source of truth for the AgentFunctionApp HTTP route prefix and HITL URLs.
+
+The server endpoints (:mod:`._app`) and the in-workflow addressing helper
+(:mod:`._hitl_context`) build the same ``{prefix}/workflow/{name}/...`` URLs. Keeping
+that shape and the route prefix in one place stops the producer and consumer from
+drifting -- previously they were only kept in sync by an integration test asserting the
+two strings match -- and lets a customized ``routePrefix`` in ``host.json`` be honored
+everywhere instead of a hardcoded ``api`` that would 404 on resume.
+"""
+
+from __future__ import annotations
+
+import functools
+import json
+import logging
+import os
+from typing import Any, cast
+
+logger = logging.getLogger(__name__)
+
+# Azure Functions' default HTTP route prefix, applied when host.json does not override
+# ``extensions.http.routePrefix``.
+DEFAULT_ROUTE_PREFIX = "api"
+
+
+@functools.lru_cache(maxsize=1)
+def route_prefix() -> str:
+    """Return the app's HTTP route prefix, honoring ``host.json``.
+
+    Azure Functions prepends ``extensions.http.routePrefix`` (default ``api``) to every
+    HTTP route, and it can be customized or set to an empty string. That value is not
+    exposed through an environment variable, so it is read from ``host.json`` under the
+    script root (``AzureWebJobsScriptRoot``, falling back to the current working
+    directory) and cached for the process. Any failure to locate or parse the file falls
+    back to the ``api`` default. Tests that vary ``host.json`` call ``route_prefix.cache_clear()``.
+    """
+    return _read_route_prefix()
+
+
+def _read_route_prefix() -> str:
+    # AzureWebJobsScriptRoot is the host-set path to the app root (mixed case is the real
+    # variable name and is case-sensitive on Linux, so it must not be upper-cased).
+    script_root = os.environ.get("AzureWebJobsScriptRoot") or os.getcwd()  # noqa: SIM112
+    host_json_path = os.path.join(script_root, "host.json")
+    try:
+        with open(host_json_path, encoding="utf-8") as f:
+            loaded = json.load(f)
+    except (OSError, ValueError):
+        logger.debug("Could not read '%s'; defaulting route prefix to '%s'.", host_json_path, DEFAULT_ROUTE_PREFIX)
+        return DEFAULT_ROUTE_PREFIX
+    if not isinstance(loaded, dict):
+        return DEFAULT_ROUTE_PREFIX
+    extensions = cast("dict[str, Any]", loaded).get("extensions")
+    if not isinstance(extensions, dict):
+        return DEFAULT_ROUTE_PREFIX
+    http = cast("dict[str, Any]", extensions).get("http")
+    if not isinstance(http, dict):
+        return DEFAULT_ROUTE_PREFIX
+    prefix = cast("dict[str, Any]", http).get("routePrefix")
+    return prefix.strip("/") if isinstance(prefix, str) else DEFAULT_ROUTE_PREFIX
+
+
+def _prefix_segment(prefix: str | None) -> str:
+    """Return the route-prefix path segment with a trailing slash, or ``""`` when empty."""
+    resolved = route_prefix() if prefix is None else prefix.strip("/")
+    return f"{resolved}/" if resolved else ""
+
+
+def build_workflow_respond_url(
+    base_url: str,
+    workflow_name: str,
+    instance_id: str,
+    request_id: str,
+    *,
+    prefix: str | None = None,
+) -> str:
+    """Build the canonical HITL respond URL a reviewer POSTs to.
+
+    ``{base}/{prefix}/workflow/{name}/respond/{instanceId}/{requestId}``. When ``prefix``
+    is omitted it is resolved from ``host.json``. ``request_id`` may be a literal
+    ``{requestId}`` placeholder to produce the templated form the run endpoint returns.
+    """
+    return f"{base_url}/{_prefix_segment(prefix)}workflow/{workflow_name}/respond/{instance_id}/{request_id}"
+
+
+def build_workflow_status_url(
+    base_url: str,
+    workflow_name: str,
+    instance_id: str,
+    *,
+    prefix: str | None = None,
+) -> str:
+    """Build the workflow status URL: ``{base}/{prefix}/workflow/{name}/status/{instanceId}``."""
+    return f"{base_url}/{_prefix_segment(prefix)}workflow/{workflow_name}/status/{instance_id}"
+
+
+def strip_route_prefix(request_url: str, *, prefix: str | None = None) -> str:
+    """Return the base URL (scheme + host) from a full request URL.
+
+    Splits the incoming request URL on the route-prefix segment (``/{prefix}/``) so it
+    honors a customized ``routePrefix``. When the prefix is empty the routes live directly
+    under the host, so it splits on the first ``/workflow/`` segment instead. Falls back to
+    the request URL without a trailing slash when neither marker is present.
+    """
+    resolved = route_prefix() if prefix is None else prefix.strip("/")
+    marker = f"/{resolved}/" if resolved else "/workflow/"
+    base_url, separator, _ = request_url.partition(marker)
+    if not separator or not base_url:
+        return request_url.rstrip("/")
+    return base_url
diff --git a/python/packages/azurefunctions/tests/test_hitl_context.py b/python/packages/azurefunctions/tests/test_hitl_context.py
index 2b88874357..0ca4466e5e 100644
--- a/python/packages/azurefunctions/tests/test_hitl_context.py
+++ b/python/packages/azurefunctions/tests/test_hitl_context.py
@@ -10,7 +10,7 @@
 import pytest
 
 from agent_framework_azurefunctions import WorkflowHitlContext
-from agent_framework_azurefunctions._hitl_context import WEBSITE_HOSTNAME_ENV
+from agent_framework_azurefunctions._hitl_context import WEBSITE_HOSTNAME_ENV, _is_loopback
 
 
 def _ctx(metadata: Any) -> SimpleNamespace:
@@ -199,3 +199,33 @@ async def test_returns_none_when_getter_absent(self) -> None:
         # A runner context that doesn't track request-info events degrades to None.
         ctx = _ctx_with_pending(None, has_getter=False)
         assert await WorkflowHitlContext.pending_request_id(ctx) is None
+
+
+class TestLoopback:
+    """Loopback detection covers the addresses ``func start`` can bind, not just localhost."""
+
+    @pytest.mark.parametrize(
+        ("host", "expected"),
+        [
+            ("localhost", True),
+            ("localhost:7071", True),
+            ("127.0.0.1", True),
+            ("127.0.0.1:7071", True),
+            ("127.5.9.9", True),
+            ("0.0.0.0", True),
+            ("0.0.0.0:7071", True),
+            ("::1", True),
+            ("[::1]:7071", True),
+            ("myapp.azurewebsites.net", False),
+            ("contoso.example.com:443", False),
+        ],
+    )
+    def test_is_loopback(self, host: str, expected: bool) -> None:
+        assert _is_loopback(host) is expected
+
+    def test_ipv6_loopback_base_url_gets_http(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # WEBSITE_HOSTNAME may report a bracketed IPv6 loopback locally; it must resolve to http.
+        monkeypatch.setenv(WEBSITE_HOSTNAME_ENV, "[::1]:7071")
+        hitl = WorkflowHitlContext.from_context(_ctx({"instance_id": "i", "workflow_name": "wf"}))
+        assert hitl is not None
+        assert hitl.base_url == "http://[::1]:7071"
diff --git a/python/packages/azurefunctions/tests/test_routes.py b/python/packages/azurefunctions/tests/test_routes.py
new file mode 100644
index 0000000000..2757ea8e7f
--- /dev/null
+++ b/python/packages/azurefunctions/tests/test_routes.py
@@ -0,0 +1,122 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Unit tests for the shared route-prefix resolution and HITL URL builders."""
+
+# pyright: reportPrivateUsage=false
+
+import json
+from collections.abc import Iterator
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from agent_framework_azurefunctions import _routes
+
+SCRIPT_ROOT_ENV = "AzureWebJobsScriptRoot"
+
+
+@pytest.fixture(autouse=True)
+def _reset_prefix_cache() -> Iterator[None]:
+    """Keep the route-prefix cache from leaking across tests."""
+    _routes.route_prefix.cache_clear()
+    yield
+    _routes.route_prefix.cache_clear()
+
+
+def _write_host_json(directory: Path, config: dict[str, Any]) -> None:
+    (directory / "host.json").write_text(json.dumps(config), encoding="utf-8")
+
+
+class TestRoutePrefix:
+    """Resolving ``extensions.http.routePrefix`` from host.json, with an ``api`` default."""
+
+    def test_defaults_to_api_without_host_json(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv(SCRIPT_ROOT_ENV, str(tmp_path))
+        assert _routes.route_prefix() == "api"
+
+    def test_defaults_to_api_when_prefix_absent(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        _write_host_json(tmp_path, {"version": "2.0", "extensions": {"http": {}}})
+        monkeypatch.setenv(SCRIPT_ROOT_ENV, str(tmp_path))
+        assert _routes.route_prefix() == "api"
+
+    def test_reads_custom_prefix(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        _write_host_json(tmp_path, {"extensions": {"http": {"routePrefix": "gateway"}}})
+        monkeypatch.setenv(SCRIPT_ROOT_ENV, str(tmp_path))
+        assert _routes.route_prefix() == "gateway"
+
+    def test_reads_empty_prefix(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        _write_host_json(tmp_path, {"extensions": {"http": {"routePrefix": ""}}})
+        monkeypatch.setenv(SCRIPT_ROOT_ENV, str(tmp_path))
+        assert _routes.route_prefix() == ""
+
+    def test_strips_surrounding_slashes(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        _write_host_json(tmp_path, {"extensions": {"http": {"routePrefix": "/custom/"}}})
+        monkeypatch.setenv(SCRIPT_ROOT_ENV, str(tmp_path))
+        assert _routes.route_prefix() == "custom"
+
+    def test_defaults_to_api_on_malformed_json(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        (tmp_path / "host.json").write_text("{ not json", encoding="utf-8")
+        monkeypatch.setenv(SCRIPT_ROOT_ENV, str(tmp_path))
+        assert _routes.route_prefix() == "api"
+
+    def test_caches_first_read(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        _write_host_json(tmp_path, {"extensions": {"http": {"routePrefix": "one"}}})
+        monkeypatch.setenv(SCRIPT_ROOT_ENV, str(tmp_path))
+        assert _routes.route_prefix() == "one"
+        # A later host.json change is not observed within a running host.
+        _write_host_json(tmp_path, {"extensions": {"http": {"routePrefix": "two"}}})
+        assert _routes.route_prefix() == "one"
+
+
+class TestUrlBuilders:
+    """Respond/status URL shapes, parameterized by an explicit prefix."""
+
+    def test_respond_url_default_prefix(self) -> None:
+        assert (
+            _routes.build_workflow_respond_url("https://h", "wf", "i", "r", prefix="api")
+            == "https://h/api/workflow/wf/respond/i/r"
+        )
+
+    def test_respond_url_custom_prefix(self) -> None:
+        assert (
+            _routes.build_workflow_respond_url("https://h", "wf", "i", "r", prefix="gw")
+            == "https://h/gw/workflow/wf/respond/i/r"
+        )
+
+    def test_respond_url_empty_prefix(self) -> None:
+        assert (
+            _routes.build_workflow_respond_url("https://h", "wf", "i", "r", prefix="")
+            == "https://h/workflow/wf/respond/i/r"
+        )
+
+    def test_respond_url_template_placeholder(self) -> None:
+        assert (
+            _routes.build_workflow_respond_url("https://h", "wf", "i", "{requestId}", prefix="api")
+            == "https://h/api/workflow/wf/respond/i/{requestId}"
+        )
+
+    def test_status_url_custom_prefix(self) -> None:
+        assert (
+            _routes.build_workflow_status_url("https://h", "wf", "i", prefix="gw")
+            == "https://h/gw/workflow/wf/status/i"
+        )
+
+    def test_status_url_empty_prefix(self) -> None:
+        assert _routes.build_workflow_status_url("https://h", "wf", "i", prefix="") == "https://h/workflow/wf/status/i"
+
+
+class TestStripRoutePrefix:
+    """Deriving the base URL from a request URL honors the configured prefix."""
+
+    def test_default_prefix(self) -> None:
+        assert _routes.strip_route_prefix("https://h:7071/api/workflow/wf/run", prefix="api") == "https://h:7071"
+
+    def test_custom_prefix(self) -> None:
+        assert _routes.strip_route_prefix("https://h/gw/workflow/wf/status/i", prefix="gw") == "https://h"
+
+    def test_empty_prefix_splits_on_workflow(self) -> None:
+        assert _routes.strip_route_prefix("https://h/workflow/wf/run", prefix="") == "https://h"
+
+    def test_marker_absent_falls_back_to_request_url(self) -> None:
+        assert _routes.strip_route_prefix("https://h/other/path", prefix="api") == "https://h/other/path"
diff --git a/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py b/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py
index 6b438da406..8ccb38a0d6 100644
--- a/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py
+++ b/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py
@@ -56,6 +56,11 @@
     workflow_orchestrator_name,
     workflow_scoped_executor_id,
 )
+from .runner_context import (
+    HOST_METADATA_INSTANCE_ID,
+    HOST_METADATA_REQUEST_PATH_PREFIX,
+    HOST_METADATA_WORKFLOW_NAME,
+)
 from .serialization import (
     SUBWORKFLOW_ADDRESS_KEY,
     SUBWORKFLOW_INPUT_KEY,
@@ -287,9 +292,9 @@ def _prepare_activity_task(
         # root down to this workflow level. For a top-level workflow the prefix is empty,
         # so this reduces to addressing the instance directly.
         "host_context": {
-            "instance_id": address["root_instance_id"],
-            "workflow_name": address["root_workflow_name"],
-            "request_path_prefix": address["request_path_prefix"],
+            HOST_METADATA_INSTANCE_ID: address["root_instance_id"],
+            HOST_METADATA_WORKFLOW_NAME: address["root_workflow_name"],
+            HOST_METADATA_REQUEST_PATH_PREFIX: address["request_path_prefix"],
         },
     }
     activity_input_json = json.dumps(activity_input)
diff --git a/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py b/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py
index eae2f117d7..d364616632 100644
--- a/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py
+++ b/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py
@@ -23,6 +23,16 @@
 from agent_framework._workflows._runner_context import YieldOutputClassifier, YieldOutputEventType
 from agent_framework._workflows._state import State
 
+# Keys of the host-metadata mapping the durable host attaches to each executor (as the
+# activity's ``host_context`` input, surfaced here via ``set_host_metadata``). These form
+# a cross-package contract: the orchestrator writes them and
+# ``WorkflowHitlContext.from_context`` in agent-framework-azurefunctions reads them back by
+# the same names, so they live here as shared constants to keep producer and consumer from
+# drifting silently (a rename would just make HITL URLs stop being built, with no error).
+HOST_METADATA_INSTANCE_ID = "instance_id"
+HOST_METADATA_WORKFLOW_NAME = "workflow_name"
+HOST_METADATA_REQUEST_PATH_PREFIX = "request_path_prefix"
+
 
 class CapturingRunnerContext(RunnerContext):
     """A RunnerContext that captures messages and events for durable activities.

From 645870e4c6b887b4491fd411034be0388b4a0e5d Mon Sep 17 00:00:00 2001
From: Ahmed Muhsin <ahmedmuhsin@microsoft.com>
Date: Wed, 15 Jul 2026 14:11:16 -0500
Subject: [PATCH 6/8] refactor(azurefunctions): derive server-side HITL URLs
 from the request URL

The run and status endpoints now derive the base URL and route prefix from the incoming request URL (the value the host actually routed) via split_request_url, so the caller-visible respond/status URLs no longer depend on reading host.json on the server. The in-workflow helper keeps reading host.json since it has no request context. Replaces strip_route_prefix and updates its tests.
---
 .../agent_framework_azurefunctions/_app.py    | 18 ++++-----
 .../agent_framework_azurefunctions/_routes.py | 39 +++++++++++--------
 .../azurefunctions/tests/test_routes.py       | 22 +++++++----
 3 files changed, 44 insertions(+), 35 deletions(-)

diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py
index 868b11423b..cbb337f650 100644
--- a/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_app.py
@@ -57,7 +57,7 @@
 from ._entities import create_agent_entity
 from ._errors import IncomingRequestError
 from ._orchestration import AgentOrchestrationContextType, AgentTask, AzureFunctionsAgentExecutor
-from ._routes import build_workflow_respond_url, build_workflow_status_url, strip_route_prefix
+from ._routes import build_workflow_respond_url, build_workflow_status_url, split_request_url
 from ._workflow import run_workflow_orchestrator
 
 logger = logging.getLogger("agent_framework.azurefunctions")
@@ -507,14 +507,16 @@ async def start_workflow_orchestration(
 
             instance_id = await client.start_new(orchestrator_name, client_input=client_input)
 
-            base_url = self._build_base_url(req.url)
-            status_url = build_workflow_status_url(base_url, workflow_name, instance_id)
+            base_url, route_prefix = split_request_url(req.url)
+            status_url = build_workflow_status_url(base_url, workflow_name, instance_id, prefix=route_prefix)
 
             return func.HttpResponse(
                 json.dumps({
                     "instanceId": instance_id,
                     "statusQueryGetUri": status_url,
-                    "respondUri": build_workflow_respond_url(base_url, workflow_name, instance_id, "{requestId}"),
+                    "respondUri": build_workflow_respond_url(
+                        base_url, workflow_name, instance_id, "{requestId}", prefix=route_prefix
+                    ),
                     "message": "Workflow started",
                 }),
                 status_code=202,
@@ -567,7 +569,7 @@ async def get_workflow_status(
             if isinstance(custom_status, dict):
                 gathered = await self._gather_pending_hitl_requests(client, cast("dict[str, Any]", custom_status))
                 if gathered:
-                    base_url = self._build_base_url(req.url)
+                    base_url, route_prefix = split_request_url(req.url)
                     pending_requests: list[dict[str, Any]] = [
                         {
                             "requestId": qualified_id,
@@ -576,7 +578,7 @@ async def get_workflow_status(
                             "requestType": req_data.get("request_type"),
                             "responseType": req_data.get("response_type"),
                             "respondUrl": build_workflow_respond_url(
-                                base_url, workflow_name, instance_id, qualified_id
+                                base_url, workflow_name, instance_id, qualified_id, prefix=route_prefix
                             ),
                         }
                         for qualified_id, req_data in gathered
@@ -741,10 +743,6 @@ async def _resolve_hitl_target(
             return None
         return await self._resolve_hitl_target(client, child_instance_id, remainder)
 
-    def _build_base_url(self, request_url: str) -> str:
-        """Extract the base URL (scheme + host) from a request URL, honoring the route prefix."""
-        return strip_route_prefix(request_url)
-
     def _is_owned_orchestration(self, status: Any, workflow_name: str) -> bool:
         """Return whether a durable orchestration status belongs to the named workflow.
 
diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_routes.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_routes.py
index 91c0a6394a..7bdcbdec3f 100644
--- a/python/packages/azurefunctions/agent_framework_azurefunctions/_routes.py
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_routes.py
@@ -3,11 +3,13 @@
 """Single source of truth for the AgentFunctionApp HTTP route prefix and HITL URLs.
 
 The server endpoints (:mod:`._app`) and the in-workflow addressing helper
-(:mod:`._hitl_context`) build the same ``{prefix}/workflow/{name}/...`` URLs. Keeping
-that shape and the route prefix in one place stops the producer and consumer from
-drifting -- previously they were only kept in sync by an integration test asserting the
-two strings match -- and lets a customized ``routePrefix`` in ``host.json`` be honored
-everywhere instead of a hardcoded ``api`` that would 404 on resume.
+(:mod:`._hitl_context`) build the same ``{prefix}/workflow/{name}/...`` URLs. Keeping the
+shape and the prefix logic here stops the two sides from drifting -- previously they were
+only kept in sync by an integration test asserting the two strings match -- and lets a
+customized ``routePrefix`` be honored instead of a hardcoded ``api`` that would 404 on
+resume. The server derives the prefix from the incoming request URL (the value the host
+actually routed); the helper, which runs inside an executor with no request context,
+reads it from ``host.json``.
 """
 
 from __future__ import annotations
@@ -17,6 +19,7 @@
 import logging
 import os
 from typing import Any, cast
+from urllib.parse import urlsplit
 
 logger = logging.getLogger(__name__)
 
@@ -96,17 +99,19 @@ def build_workflow_status_url(
     return f"{base_url}/{_prefix_segment(prefix)}workflow/{workflow_name}/status/{instance_id}"
 
 
-def strip_route_prefix(request_url: str, *, prefix: str | None = None) -> str:
-    """Return the base URL (scheme + host) from a full request URL.
+def split_request_url(request_url: str) -> tuple[str, str]:
+    """Return ``(base_url, route_prefix)`` derived from an incoming request URL.
 
-    Splits the incoming request URL on the route-prefix segment (``/{prefix}/``) so it
-    honors a customized ``routePrefix``. When the prefix is empty the routes live directly
-    under the host, so it splits on the first ``/workflow/`` segment instead. Falls back to
-    the request URL without a trailing slash when neither marker is present.
+    On the server the request URL is the authoritative source for the prefix, since the
+    host served it through the configured ``routePrefix``. The scheme and host form the
+    base URL, and the path before the first ``/workflow/`` segment is the prefix (empty
+    when the routes sit directly under the host). Falls back to ``(request_url, "")`` when
+    the value is not an absolute URL.
     """
-    resolved = route_prefix() if prefix is None else prefix.strip("/")
-    marker = f"/{resolved}/" if resolved else "/workflow/"
-    base_url, separator, _ = request_url.partition(marker)
-    if not separator or not base_url:
-        return request_url.rstrip("/")
-    return base_url
+    parts = urlsplit(request_url)
+    if not (parts.scheme and parts.netloc):
+        return request_url.rstrip("/"), ""
+    base_url = f"{parts.scheme}://{parts.netloc}"
+    index = parts.path.find("/workflow/")
+    prefix = parts.path[:index].strip("/") if index != -1 else ""
+    return base_url, prefix
diff --git a/python/packages/azurefunctions/tests/test_routes.py b/python/packages/azurefunctions/tests/test_routes.py
index 2757ea8e7f..f00e989326 100644
--- a/python/packages/azurefunctions/tests/test_routes.py
+++ b/python/packages/azurefunctions/tests/test_routes.py
@@ -106,17 +106,23 @@ def test_status_url_empty_prefix(self) -> None:
         assert _routes.build_workflow_status_url("https://h", "wf", "i", prefix="") == "https://h/workflow/wf/status/i"
 
 
-class TestStripRoutePrefix:
-    """Deriving the base URL from a request URL honors the configured prefix."""
+class TestSplitRequestUrl:
+    """Deriving base URL and route prefix from an incoming request URL."""
 
     def test_default_prefix(self) -> None:
-        assert _routes.strip_route_prefix("https://h:7071/api/workflow/wf/run", prefix="api") == "https://h:7071"
+        assert _routes.split_request_url("https://h:7071/api/workflow/wf/run") == ("https://h:7071", "api")
 
     def test_custom_prefix(self) -> None:
-        assert _routes.strip_route_prefix("https://h/gw/workflow/wf/status/i", prefix="gw") == "https://h"
+        assert _routes.split_request_url("https://h/gw/workflow/wf/status/i") == ("https://h", "gw")
 
-    def test_empty_prefix_splits_on_workflow(self) -> None:
-        assert _routes.strip_route_prefix("https://h/workflow/wf/run", prefix="") == "https://h"
+    def test_empty_prefix(self) -> None:
+        assert _routes.split_request_url("https://h/workflow/wf/run") == ("https://h", "")
 
-    def test_marker_absent_falls_back_to_request_url(self) -> None:
-        assert _routes.strip_route_prefix("https://h/other/path", prefix="api") == "https://h/other/path"
+    def test_multi_segment_prefix(self) -> None:
+        assert _routes.split_request_url("https://h/a/b/workflow/wf/run") == ("https://h", "a/b")
+
+    def test_no_workflow_segment(self) -> None:
+        assert _routes.split_request_url("https://h/api/health") == ("https://h", "")
+
+    def test_non_absolute_falls_back(self) -> None:
+        assert _routes.split_request_url("/api/workflow/wf/run") == ("/api/workflow/wf/run", "")

From 82ed8637a5fdc5d981dd3451ce2292a77dfae110 Mon Sep 17 00:00:00 2001
From: Ahmed Muhsin <ahmedmuhsin@microsoft.com>
Date: Wed, 15 Jul 2026 14:30:05 -0500
Subject: [PATCH 7/8] test(durabletask): enforce sub-workflow ordinal and
 read-index agreement

Extract the read-side subworkflows grouping into a shared _index_subworkflows helper (used by the orchestrator) and add test_readside_index_matches_dispatch_ordinal, which round-trips a fan-out through that helper and asserts subworkflows[executor][ordinal] resolves to the child the dispatch stamped that ordinal onto. Turns the previously comment-only write-ordinal / read-index invariant into a shared, CI-enforced one.
---
 .../_workflows/orchestrator.py                | 31 +++++++++++++------
 .../tests/test_subworkflow_orchestration.py   | 28 ++++++++++++++---
 2 files changed, 45 insertions(+), 14 deletions(-)

diff --git a/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py b/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py
index 8ccb38a0d6..346d6a3985 100644
--- a/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py
+++ b/python/packages/durabletask/agent_framework_durabletask/_workflows/orchestrator.py
@@ -961,6 +961,24 @@ def _prepare_all_tasks(
     return all_tasks, task_metadata_list, remaining_agent_messages
 
 
+def _index_subworkflows(task_metadata_list: list[TaskMetadata]) -> dict[str, list[str]]:
+    """Group dispatched sub-workflow child instance ids by executor id, in dispatch order.
+
+    This is the read-side addressing map the parent publishes to its custom status so the
+    status/respond endpoints can resolve a nested pending request: a request qualified as
+    ``{executorId}~{ordinal}~{bare}`` maps to ``subworkflows[executorId][ordinal]``. That
+    ordinal is the child's position in this list, which must equal the write-side ordinal
+    :func:`_prepare_all_tasks` stamps into the child's request-path prefix. Both derive from
+    the same ``task_metadata_list`` order, so building the map here in one place keeps the
+    two sides from drifting (guarded by ``test_readside_index_matches_dispatch_ordinal``).
+    """
+    subworkflows: dict[str, list[str]] = {}
+    for meta in task_metadata_list:
+        if meta.task_type == TaskType.SUBWORKFLOW and meta.child_instance_id is not None:
+            subworkflows.setdefault(meta.executor_id, []).append(meta.child_instance_id)
+    return subworkflows
+
+
 # ============================================================================
 # Main Orchestrator
 # ============================================================================
@@ -1111,15 +1129,10 @@ def publish_live_status(
             logger.debug("Executing %d tasks in parallel (agents + activities)", len(all_tasks))
             # Record dispatched sub-workflow child instance ids before suspending in
             # task_all. While a nested sub-workflow waits for human input, this parent
-            # stays suspended here, so its custom status must already carry the child
-            # ids for the read side to discover and qualify nested pending requests.
-            # Grouped as {executorId: [childInstanceId, ...]} in dispatch order so a
-            # node that dispatches several children this superstep keeps each one
-            # addressable by its ordinal.
-            active_subworkflows: dict[str, list[str]] = {}
-            for meta in task_metadata_list:
-                if meta.task_type == TaskType.SUBWORKFLOW and meta.child_instance_id is not None:
-                    active_subworkflows.setdefault(meta.executor_id, []).append(meta.child_instance_id)
+            # stays suspended here, so its custom status must already carry the child ids
+            # for the read side to discover and qualify nested pending requests (see
+            # _index_subworkflows for the dispatch-order / ordinal addressing contract).
+            active_subworkflows = _index_subworkflows(task_metadata_list)
             if active_subworkflows:
                 publish_live_status("running", subworkflows=active_subworkflows)
             raw_results = yield ctx.task_all(all_tasks)
diff --git a/python/packages/durabletask/tests/test_subworkflow_orchestration.py b/python/packages/durabletask/tests/test_subworkflow_orchestration.py
index 588f5317db..3743396562 100644
--- a/python/packages/durabletask/tests/test_subworkflow_orchestration.py
+++ b/python/packages/durabletask/tests/test_subworkflow_orchestration.py
@@ -22,8 +22,10 @@
 from agent_framework_durabletask._workflows.orchestrator import (
     SUBWORKFLOW_ADDRESS_KEY,
     SUBWORKFLOW_INPUT_KEY,
+    TaskMetadata,
     TaskType,
     _coerce_initial_input,
+    _index_subworkflows,
     _prepare_all_tasks,
     _prepare_subworkflow_task,
     _process_subworkflow_result,
@@ -312,10 +314,12 @@ class TestSubworkflowAddressPropagation:
     for the tricky case: one node fanning out to several children in one superstep.
     """
 
-    def _dispatch(self, address: dict[str, str], message_count: int) -> tuple[list[dict[str, str]], list[str]]:
+    def _dispatch(
+        self, address: dict[str, str], message_count: int
+    ) -> tuple[list[dict[str, str]], list[str], list[TaskMetadata]]:
         """Run _prepare_all_tasks for one WorkflowExecutor node fanning out N children.
 
-        Returns (child_addresses, child_instance_ids) in dispatch order.
+        Returns (child_addresses, child_instance_ids, task_metadata) in dispatch order.
         """
         node_id = "review_sub"
         executor = _subworkflow_executor(node_id, "human_review")
@@ -337,11 +341,11 @@ def _call_sub(name: str, input_: dict[str, object], *, instance_id: str) -> str:
         _all_tasks, task_metadata, _remaining = _prepare_all_tasks(ctx, workflow, pending, None, [0], address)
         child_ids = [m.child_instance_id for m in task_metadata if m.task_type == TaskType.SUBWORKFLOW]
         assert all(cid is not None for cid in child_ids)
-        return captured, [cid for cid in child_ids if cid is not None]
+        return captured, [cid for cid in child_ids if cid is not None], task_metadata
 
     def test_fanout_prefixes_match_readside_qualification(self) -> None:
         top = {"root_instance_id": "root", "root_workflow_name": "moderation_pipeline", "request_path_prefix": ""}
-        addresses, child_ids = self._dispatch(top, message_count=3)
+        addresses, child_ids, _task_metadata = self._dispatch(top, message_count=3)
 
         # Read side: subworkflows["review_sub"] = [child0, child1, child2]; a nested
         # request from child ``ordinal`` is qualified as review_sub~{ordinal}~{bare}.
@@ -364,9 +368,23 @@ def test_prefix_accumulates_when_already_nested(self) -> None:
             "root_workflow_name": "moderation_pipeline",
             "request_path_prefix": "outer_node~2~",
         }
-        addresses, _child_ids = self._dispatch(nested, message_count=2)
+        addresses, _child_ids, _task_metadata = self._dispatch(nested, message_count=2)
 
         assert [a["request_path_prefix"] for a in addresses] == [
             "outer_node~2~review_sub~0~",
             "outer_node~2~review_sub~1~",
         ]
+
+    def test_readside_index_matches_dispatch_ordinal(self) -> None:
+        # Close the loop through the read-side map the parent actually publishes: a nested
+        # request qualified review_sub~{ordinal}~ must resolve, via subworkflows[executor],
+        # to the same child the dispatch stamped that ordinal onto. Guards the write ordinal
+        # and read index from drifting if task_metadata order or the grouping ever changes.
+        top = {"root_instance_id": "root", "root_workflow_name": "moderation_pipeline", "request_path_prefix": ""}
+        addresses, child_ids, task_metadata = self._dispatch(top, message_count=3)
+
+        subworkflows = _index_subworkflows(task_metadata)
+        assert subworkflows == {"review_sub": child_ids}
+        for ordinal, (child_id, child_address) in enumerate(zip(child_ids, addresses, strict=True)):
+            assert child_address["request_path_prefix"] == qualify_subworkflow_request_id("review_sub", ordinal, "")
+            assert subworkflows["review_sub"][ordinal] == child_id

From 2bbaa212fd591d6997fc5d71e9786f6726a73055 Mon Sep 17 00:00:00 2001
From: Ahmed Muhsin <ahmedmuhsin@microsoft.com>
Date: Wed, 15 Jul 2026 16:24:10 -0500
Subject: [PATCH 8/8] fix(azurefunctions): suppress bandit B104 on loopback
 host set

---
 .../agent_framework_azurefunctions/_hitl_context.py             | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
index ecc344ba02..d14a59e77c 100644
--- a/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_hitl_context.py
@@ -44,7 +44,7 @@
 
 # Loopback hosts that resolve to ``http`` (not ``https``) when WEBSITE_HOSTNAME is
 # host-only; covers the addresses ``func start`` can bind locally.
-_LOOPBACK_HOSTS = frozenset({"localhost", "0.0.0.0", "::1"})  # noqa: S104
+_LOOPBACK_HOSTS = frozenset({"localhost", "0.0.0.0", "::1"})  # noqa: S104  # nosec B104
 
 
 def _is_loopback(host: str) -> bool: