From fda0d2deef50fbfb29d509946afe1eb9589a880c 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/4] 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 9b3d3db27af..2976a38858d 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 00000000000..be48cea35fc
--- /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 aec5e6615e2..2778c2cb281 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 1b94eba3b35..c7174fa4621 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 00000000000..f6ee6e10d37
--- /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 ba6a0e996d3..565bb51b9e9 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 7a914cee51d..11578758451 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 7fbdfc95f2d..9fa8b642c67 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 d8ada5b5933..6b438da406e 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 1851339bb37..eae2f117d77 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 680a614c6b7..cabd2c4251b 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 8ec735cdcd1..850336ac1f1 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 b2ff9c159bc..4e6a8aca729 100644
--- a/python/packages/durabletask/tests/test_workflow_activity.py
+++ b/python/packages/durabletask/tests/test_workflow_activity.py
@@ -117,6 +117,42 @@ async def mutate(message: Any, source_executor_ids: Any, state: Any, runner_cont
         assert "keep" not in result["shared_state_deletes"]
 
 
+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 80295c9c804..381839b2c45 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 e51f2e98a87..499d66779be 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 bea2887f907..a35a88dafc1 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 7cb370517a78e8bec5ad00d71d7c99079d96ba61 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/4] 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 be48cea35fc..f0daea2577e 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 f6ee6e10d37..2b888743570 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 499d66779be..37ea8fdf1a4 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 a35a88dafc1..1b6e0f9aa48 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 57b12c3f81b3c1c6fbe4d3c124fa7f384ada7d5f 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/4] 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 f0daea2577e..59a8939aab0 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 60e575ad829..30acd3f686a 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 cdbbe9b3689..e1bce747298 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 a1030c7c422e1d3173c13f70d1255aa27ada6694 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/4] 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 2778c2cb281..f018f07e86f 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 c7174fa4621..70d0a856756 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 850336ac1f1..588f5317dbd 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 30acd3f686a..d7fa9ca7351 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 e1bce747298..1cbfb7770d1 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.