Merge branch 'main' into feature/python-subworkflow-checkpoint

Add warning
Python: Improve python package management operations (#7274 )
2026-07-24 12:25:13 +09:00 · 2026-07-23 15:25:58 -07:00 · 2026-07-23 22:14:10 +00:00 · 2026-07-23 21:27:26 +00:00 · 2026-07-20 14:14:24 -07:00 · 2026-07-20 13:27:35 -07:00
20 changed files with 1608 additions and 284 deletions
@@ -28,7 +28,7 @@ For release work, derive the live tier map at release time from `python/PACKAGE_

 ## Inputs to confirm before bumping

-1. **The changeset**: explicit commits/PRs the release covers, OR derive from `git log ${LAST_RELEASED_TAG}..origin/main -- python/`.
+1. **The changeset**: explicit commits/PRs the release covers, OR derive from `git log ${LAST_RELEASED_TAG}..${RELEASE_BASE} -- python/`.
 2. **Per-package CHANGELOG entries**: which packages will get a line in the new release section. This list IS the bump list.
 3. **Per-released-package semver bump**: for each released-tier package that has a CHANGELOG entry, decide PATCH / MINOR / MAJOR.
 4. **Date stamp** (only if any alpha/beta is being bumped): default from the `python-package-management`
@@ -55,12 +55,20 @@ If the user states target versions or a date explicitly, use exactly what they s
 git fetch origin main --tags --quiet
 git fetch upstream main --tags --quiet 2>/dev/null || true
 git status
+
+# Fork clones use upstream/main as the authoritative release base; direct clones use origin/main.
+if git show-ref --verify --quiet refs/remotes/upstream/main; then
+  RELEASE_BASE=upstream/main
+else
+  RELEASE_BASE=origin/main
+fi
+git log -1 --oneline "$RELEASE_BASE"
 ```

 If the user already has a `bump-py-ver-release-*` branch checked out, use it. Otherwise:

 ```bash
-git checkout -b bump-py-ver-release-YYMMDD origin/main
+git checkout -b bump-py-ver-release-YYMMDD "$RELEASE_BASE"
 ```

 ### 2. Build the live tier map
@@ -84,20 +92,20 @@ echo "Compare base: $LAST_RELEASED_TAG"
 List commits and packages touched:

 ```bash
-git log --oneline ${LAST_RELEASED_TAG}..origin/main -- python/ ':!python/CHANGELOG.md'
+git log --oneline ${LAST_RELEASED_TAG}..${RELEASE_BASE} -- python/ ':!python/CHANGELOG.md'

 # Per-commit package footprint
-for sha in $(git log --format='%H' ${LAST_RELEASED_TAG}..origin/main -- python/); do
+for sha in $(git log --format='%H' ${LAST_RELEASED_TAG}..${RELEASE_BASE} -- python/); do
  echo "--- $(git show -s --format='%h %s' $sha) ---"
  git show --name-only --format='' $sha | grep '^python/packages/' | \
    sed 's|^python/packages/||' | awk -F/ '{print $1}' | sort -u
 done
 ```

-If the release ultimately tags from `upstream/main` but `origin/main` is behind, also run:
+When both remotes exist, record whether the fork is behind the authoritative base:

 ```bash
-git log --oneline ${LAST_RELEASED_TAG}..upstream/main -- python/ ':!python/CHANGELOG.md'
+git rev-list --left-right --count origin/main...upstream/main
 ```

 If user provides an explicit commit/PR list, treat THAT as authoritative.
@@ -108,14 +116,14 @@ Aggregate the per-commit footprint into a single union across the whole range. T

 ```bash
 # Union of all touched package directories across the range
-git log --name-only --format='' ${LAST_RELEASED_TAG}..origin/main -- python/packages/ \
+git log --name-only --format='' ${LAST_RELEASED_TAG}..${RELEASE_BASE} -- python/packages/ \
  | grep '^python/packages/' \
  | sed 's|^python/packages/||' \
  | awk -F/ '{print $1}' \
  | sort -u

 # Root-level files (drive a root agent-framework entry if substantive)
-git log --name-only --format='' ${LAST_RELEASED_TAG}..origin/main \
+git log --name-only --format='' ${LAST_RELEASED_TAG}..${RELEASE_BASE} \
  -- python/pyproject.toml python/agent_framework_meta/ python/README.md \
  2>/dev/null | grep -v '^$' | sort -u
 ```
@@ -175,13 +183,13 @@ Before moving on, prove that every ship-affecting touched package has at least o

 ```bash
 # 1. Touched ship-affecting packages and root package files (from step 3a)
-TOUCHED_PACKAGES=$(git log --name-only --format='' ${LAST_RELEASED_TAG}..origin/main -- python/packages/ \
+TOUCHED_PACKAGES=$(git log --name-only --format='' ${LAST_RELEASED_TAG}..${RELEASE_BASE} -- python/packages/ \
  | grep '^python/packages/' \
  | sed 's|^python/packages/||' \
  | awk -F/ '{print $1}' \
  | sort -u)

-ROOT_TOUCHED=$(git log --name-only --format='' ${LAST_RELEASED_TAG}..origin/main \
+ROOT_TOUCHED=$(git log --name-only --format='' ${LAST_RELEASED_TAG}..${RELEASE_BASE} \
  -- python/pyproject.toml python/agent_framework_meta/ python/README.md \
  2>/dev/null | grep -v '^$' | sort -u)

@@ -279,7 +287,7 @@ Spot-check with `grep '^version' python/pyproject.toml python/packages/*/pyproje
 Only relevant when `core` itself bumped this cycle. Two policies, pick one explicitly with the user:

 - **Conservative (default)**: raise `agent-framework-core>=X.Y.Z` to the new core version on every non-core package that is ALSO bumping this cycle. Leaves packages-not-bumped at their existing floor.
- **Strict per-upstream-doc**: only raise the floor on packages that actually consume a new core API introduced in the bump. This requires per-package code inspection. Use only when the user is comfortable letting `validate-dependency-bounds-test` (lower-resolution pass) catch any mistakes.
+- **Strict per-upstream-doc**: only raise the floor on packages that actually consume a new core API introduced in the bump. This requires per-package code inspection because release probes use the co-released local core and cannot prove compatibility with an older published core floor.

 When raising a core floor, replace only the `>=OLD` half of the bound you intend to change:

@@ -294,12 +302,29 @@ If `core` did not bump this cycle, do not touch floors.
 ### 7. Validate

 ```bash
-cd python && uv run poe validate-dependency-bounds-test
+cd python && uv run poe validate-python-release --base-ref "$RELEASE_BASE"
 ```

-Must exit 0. This is the safety net for selective bumping: the lower-resolution pass catches floors set too low for code that depends on new APIs, and the upper pass catches caps that exclude installable versions. If it fails, the output names the offending bound — fix and re-run before committing. This step also regenerates `uv.lock` to match new bounds.
+Use the same freshly fetched main ref that the release branch was based on (`upstream/main` above; use `origin/main`
+when that is the authoritative release base). Must exit 0. This task first regenerates `uv.lock`, then discovers the
+package `pyproject.toml` files changed from that base and runs their published runtime dependencies and
+non-development extras through lock-independent `lowest-direct` and `highest` import probes. The probes run in
+parallel, derive the minimum supported Python minor from each package's internal editable closure, and share a hard
+300-second deadline. Use `--python` only when the release requires an explicit interpreter override.

-If only prereleases changed (no `core` bump, no floor changes), this validation is still required — `uv.lock` regeneration alone justifies the run.
+This is the release safety net for selective bumping: the lower probe catches unresolvable or unimportable external
+floors, internal constraints that reject co-released package versions, and the upper probe catches caps that exclude
+an installable package set. The JSON report records the concrete versions resolved in both scenarios. It does not
+replace the package-by-package code inspection required by the strict core-floor policy. If it fails, fix the named
+package/bound and re-run before committing.
+
+Do not substitute the workspace-wide `validate-dependency-bounds-test` command here. That command runs every
+package's full tests and Pyright in separate isolated environments and is intentionally reserved for CI or an
+explicit dependency-range audit. If the release itself changes an external dependency range, also run
+`validate-dependency-bounds-project --mode both --package <pkg> --dependency <name>` for that dependency.
+
+If only prereleases changed (no `core` bump, no floor changes), release validation is still required because the
+lockfile and both ends of each changed package's published dependency metadata must remain installable.

 ### 8. Commit (expect hook retry)

@@ -349,7 +374,7 @@ The push output includes a `Create a pull request for '<branch>' on GitHub by vi
  do not infer a local timezone from the user's current shell.
 - **`Co-Authored-By` trailer.** Never add it. Rewrite/amend if it slipped in.
 - **Stale inventory in this skill.** Always read `python/PACKAGE_STATUS.md` for the live tier map. Do not trust a hardcoded list.
- **Divergent origin vs upstream.** If the release tags from `upstream/main` but `origin/main` is behind, check both — warn if they differ and offer to sync.
+- **Divergent origin vs upstream.** In fork clones, use freshly fetched `upstream/main` consistently for branch creation, changeset discovery, and release validation. A stale `origin/main` must never become the implicit compare base.
 - **`--pre` README cleanup on promotion.** When a package is promoted to `released` in this cycle, grep for `pip install agent-framework-<pkg> --pre` in READMEs and drop the `--pre` flag.
 - **RC counter inflation.** Do not increment `1.0.0rcN` without a CHANGELOG entry for that package. The counter tracks iterations, not calendar.

@@ -357,5 +382,6 @@ The push output includes a `Create a pull request for '<branch>' on GitHub by vi

 - Package lifecycle and versioning source of truth: `python/.github/skills/python-package-management/SKILL.md`
 - Lifecycle source of truth: `python/PACKAGE_STATUS.md`
- Validator: `python/scripts/dependencies/validate_dependency_bounds.py` (runs `lowest-direct` and `highest` resolution smoke tests; catches floors/caps that don't match the code)
+- Release validator: `python/scripts/dependencies/validate_dependency_bounds.py --mode release` (changed-package,
+  lock-independent `lowest-direct` and `highest` import probes under a five-minute deadline)
 - Poe task definitions: `python/pyproject.toml` `[tool.poe.tasks]`
@@ -45,9 +45,13 @@ uv lock --upgrade-package <dependency-name> && uv run poe install
 # Refresh exact development dependency-group pins, lockfile, and validation in one run
 uv run poe upgrade-dev-dependencies

-# First, run workspace-wide lower/upper compatibility gates
+# Release cuts: refresh uv.lock and probe changed packages at both bound extremes.
+# The release probe has a shared five-minute deadline.
+uv run poe validate-python-release --base-ref upstream/main
+
+# Exhaustive test+typing matrix (slow; use for deliberate dependency-range work or CI)
 uv run poe validate-dependency-bounds-test
-# Defaults to --package "*"; pass a package to scope test mode
+# Defaults to --package "*"; scope locally whenever possible.
 uv run poe validate-dependency-bounds-test --package core

 # Then expand bounds for one dependency in the target package
@@ -66,7 +70,16 @@ uv run poe add-dependency-and-validate-bounds --package core --dependency "<depe
 - Prerelease (`dev`/`a`/`b`/`rc`) and `<1.0` dependencies should use hard bounds with an explicit upper cap (avoid open-ended ranges).
 - For `<1.0` dependencies, prefer the broadest validated range the package can really support. That may be a patch line, a minor line, or multiple minor lines when checks/tests show the broader lane is compatible.
 - Prefer supporting multiple majors when practical; if APIs diverge across supported majors, use version-conditional imports/paths.
- For dependency changes, run workspace-wide bound gates first, then `validate-dependency-bounds-project --mode both` for the target package/dependency to keep minimum and maximum constraints current. The same task can also drive repo-wide upper-bound automation by using `--package "*"` and omitting `--dependency`.
+- For release-only version, lifecycle, pin, and internal-floor edits, use `validate-python-release`. It refreshes
+  `uv.lock`, finds changed package metadata relative to the selected main ref, and runs the changed packages'
+  published runtime dependencies and non-development extras through lock-independent `lowest-direct` and `highest`
+  import probes on the minimum Python minor supported by each package's internal editable closure. The probes run
+  concurrently under one 300-second deadline; pass `--python` only when an explicit interpreter override is needed.
+- For deliberate external dependency-range changes, use
+  `validate-dependency-bounds-project --mode both` for the target package/dependency to find and validate the actual
+  minimum and maximum constraints. Scope the exhaustive `validate-dependency-bounds-test` matrix to affected
+  packages during local iteration; reserve the workspace-wide form for CI or an intentional full audit. The same
+  project task can drive repo-wide upper-bound automation by using `--package "*"` and omitting `--dependency`.
 - Prefer targeted lock updates with `uv lock --upgrade-package <dependency-name>` to reduce `uv.lock` merge conflicts.
 - Use `add-dependency-and-validate-bounds` for package-scoped dependency additions plus bound validation in one command.
 - Keep shared tooling and source/type-check support in the root or package `dev` group. Put package-specific test
@@ -115,6 +115,18 @@ class CapturingRunnerContext(RunnerContext):
        """Checkpointing not supported in activity context."""
        raise NotImplementedError("Checkpointing is not supported in Azure Functions activity context")

+    async def build_checkpoint(
+        self,
+        workflow_name: str,
+        graph_signature_hash: str,
+        state: State,
+        previous_checkpoint_id: str | None,
+        iteration_count: int,
+        metadata: dict[str, Any] | None = None,
+    ) -> WorkflowCheckpoint:
+        """Checkpointing not supported in activity context."""
+        raise NotImplementedError("Checkpointing is not supported in Azure Functions activity context")
+
    async def load_checkpoint(self, checkpoint_id: str) -> WorkflowCheckpoint | None:
        """Checkpointing not supported in activity context."""
        raise NotImplementedError("Checkpointing is not supported in Azure Functions activity context")
@@ -2096,6 +2096,12 @@ def _collect_approval_responses(
    return fcc_todo


+def _is_approval_placeholder_result(content: Content) -> bool:
+    """Whether a function_result is the stand-in emitted while approval is pending."""
+    result = getattr(content, "result", None)
+    return isinstance(result, str) and "[APPROVAL_PENDING]" in result
+
+
 def _replace_approval_contents_with_results(
    messages: list[Message],
    fcc_todo: dict[str, Content],
@@ -2119,12 +2125,30 @@ def _replace_approval_contents_with_results(
    # Track which call_ids had their placeholders replaced
    placeholders_replaced: set[str] = set()

-    for msg in messages:
-        # First pass - collect existing function call IDs to avoid duplicates
-        existing_call_ids = {
-            content.call_id for content in msg.contents if content.type == "function_call" and content.call_id
-        }
+    # Collect *pending* function call IDs across all messages to avoid duplicates. The
+    # function call and its approval request are frequently carried in separate messages
+    # (e.g. when a hosting layer replays them as separate items on an approval round trip),
+    # so scoping this per-message would let the same call_id be restored twice and leave
+    # the copy without a result unanswered.
+    #
+    # Calls that already carry a real result are excluded: reusing a call_id for a later
+    # invocation is supported, and a completed pair must not suppress the fresh request —
+    # that would drop the new call and attach its result to the old one. Placeholder
+    # results still count as pending, since the call they answer is the one being restored.
+    answered_call_ids = {
+        content.call_id
+        for msg in messages
+        for content in msg.contents
+        if content.type == "function_result" and content.call_id and not _is_approval_placeholder_result(content)
+    }
+    existing_call_ids = {
+        content.call_id
+        for msg in messages
+        for content in msg.contents
+        if content.type == "function_call" and content.call_id and content.call_id not in answered_call_ids
+    }

+    for msg in messages:
        # Track approval requests that should be removed (duplicates)
        contents_to_remove: list[int] = []

@@ -2140,6 +2164,8 @@ def _replace_approval_contents_with_results(
                elif content.function_call is not None:
                    # Put back the function call content only if it doesn't exist
                    msg.contents[content_idx] = content.function_call
+                    if content.function_call.call_id:
+                        existing_call_ids.add(content.function_call.call_id)
            elif content.type == "function_approval_response":
                # Skip hosted tool approvals — they must pass through to the API unchanged
                if _is_hosted_tool_approval(content):
@@ -2169,12 +2195,7 @@ def _replace_approval_contents_with_results(
                    msg.role = "tool"
            elif content.type == "function_result":
                # Check if this is a placeholder result that should be replaced
-                if (
-                    hasattr(content, "result")
-                    and isinstance(content.result, str)
-                    and "[APPROVAL_PENDING]" in content.result
-                    and content.call_id in result_by_call_id
-                ):
+                if _is_approval_placeholder_result(content) and content.call_id in result_by_call_id:
                    # Replace placeholder with actual result
                    msg.contents[content_idx] = result_by_call_id[content.call_id]
                    placeholders_replaced.add(content.call_id)
@@ -811,6 +811,24 @@ class FunctionalWorkflow:
                execution is not allowed).
        """
        self._validate_run_params(message, responses, checkpoint_id)
+        # Warn (but don't block) when a fresh message or a checkpoint restore begins while a prior
+        # run left request_info events pending. Mirrors Workflow.run. Delivering responses is the
+        # normal way to complete the pending cycle and is intentionally not warned.
+        if (message is not None or checkpoint_id is not None) and self._last_pending_request_ids:
+            logger.warning(
+                "Workflow %s received %s while %d request_info event(s) are still pending from an "
+                "unfinished request/response cycle; %s. Deliver responses (responses=...) to complete "
+                "the pending cycle before starting new input.",
+                self.name,
+                "a fresh message" if message is not None else "a checkpoint restore",
+                len(self._last_pending_request_ids),
+                (
+                    "those requests remain answerable, but this run advances workflow state, so a "
+                    "response that arrives later may apply to a workflow that has moved on"
+                    if message is not None
+                    else "those pending requests will be overwritten by the checkpoint's state"
+                ),
+            )
        if responses and checkpoint_id is None:
            # Require at least one response key to match a currently-pending
            # request; prevents silent replay against stale state while still
@@ -245,7 +245,11 @@ class RunnerImpl:
        self._state.commit()

    async def create_checkpoint_if_enabled(self) -> None:
-        """Create a checkpoint if checkpointing is enabled and attach a label and metadata."""
+        """Create a checkpoint and save the checkpoint to the configured storage if one is configured.
+
+        Note:
+            1. This method has no effect if checkpointing is not enabled in the context.
+        """
        if not self._ctx.has_checkpointing():
            return

@@ -340,6 +344,57 @@ class RunnerImpl:
            logger.error(f"Failed to restore from checkpoint {checkpoint_id}: {e}")
            raise WorkflowCheckpointException(f"Failed to restore from checkpoint {checkpoint_id}") from e

+    async def build_checkpoint(self) -> WorkflowCheckpoint:
+        """Create a checkpoint object.
+
+        Returns:
+            A ``WorkflowCheckpoint``.
+        """
+        # Persist executor snapshots into committed shared state before exporting it.
+        await self._prepare_checkpoint_state()
+        return await self._ctx.build_checkpoint(
+            self._workflow_name,
+            self._graph_signature_hash,
+            self._state,
+            None,
+            self._iteration,
+        )
+
+    async def restore_checkpoint(self, checkpoint: WorkflowCheckpoint) -> None:
+        """Restore runner state from an in-memory ``WorkflowCheckpoint`` object.
+
+        Unlike :meth:`restore_from_checkpoint`, this does not load from a storage
+        backend; it applies a checkpoint the caller already holds - for example, a
+        child workflow checkpoint embedded in a parent ``WorkflowExecutor``'s state.
+
+        Restores shared state, executor snapshots, in-flight messages, and pending
+        request_info events, then marks the runner as resumed.
+
+        Args:
+            checkpoint: The checkpoint whose state should be restored.
+
+        Raises:
+            WorkflowCheckpointException: If the checkpoint's graph signature does not
+                match this runner's workflow, or if restoration otherwise fails.
+        """
+        if self._graph_signature_hash != checkpoint.graph_signature_hash:
+            raise WorkflowCheckpointException(
+                "Workflow graph has changed since the checkpoint was created. "
+                "Please rebuild the original workflow before resuming."
+            )
+
+        try:
+            # Clear first so import_state (which merges) does not leak stale keys from a
+            # prior run on this Workflow instance.
+            self._state.clear()
+            self._state.import_state(checkpoint.state)
+            await self._restore_executor_states()
+            await self._ctx.apply_checkpoint(checkpoint)
+            self._mark_resumed(checkpoint)
+        except Exception as e:
+            logger.error(f"Failed to restore from checkpoint {checkpoint.checkpoint_id}: {e}")
+            raise WorkflowCheckpointException(f"Failed to restore from checkpoint {checkpoint.checkpoint_id}") from e
+
    async def _save_executor_states(self) -> None:
        """Populate executor state by calling checkpoint hooks on executors."""
        for exec_id, executor in self._executors.items():
@@ -195,6 +195,34 @@ class RunnerContext(Protocol):
        """
        ...

+    async def build_checkpoint(
+        self,
+        workflow_name: str,
+        graph_signature_hash: str,
+        state: State,
+        previous_checkpoint_id: CheckpointID | None,
+        iteration_count: int,
+        metadata: dict[str, Any] | None = None,
+    ) -> WorkflowCheckpoint:
+        """Build a checkpoint and return it for the caller to own.
+
+        The checkpoint is constructed in memory and handed back to the caller; nothing is
+        persisted and no checkpoint storage is required.
+
+        Args:
+            workflow_name: The name of the workflow for which the checkpoint is being created.
+            graph_signature_hash: Hash of the workflow graph topology to
+                validate checkpoint compatibility during restore.
+            state: The state to include in the checkpoint.
+            previous_checkpoint_id: The ID of the previous checkpoint, if any, to form a checkpoint chain.
+            iteration_count: The current iteration count of the workflow.
+            metadata: Optional metadata to associate with the checkpoint.
+
+        Returns:
+            A ``WorkflowCheckpoint`` of the current context state.
+        """
+        ...
+
    async def create_checkpoint(
        self,
        workflow_name: str,
@@ -204,7 +232,7 @@ class RunnerContext(Protocol):
        iteration_count: int,
        metadata: dict[str, Any] | None = None,
    ) -> CheckpointID:
-        """Create a checkpoint of the current workflow state.
+        """Persist a checkpoint of the current workflow state to configured storage and return its ID.

        Args:
            workflow_name: The name of the workflow for which the checkpoint is being created.
@@ -219,6 +247,9 @@ class RunnerContext(Protocol):

        Returns:
            The ID of the created checkpoint.
+
+        Raises:
+            ValueError: If checkpoint storage is not configured.
        """
        ...

@@ -381,6 +412,27 @@ class InProcRunnerContext:
    def has_checkpointing(self) -> bool:
        return self._get_effective_checkpoint_storage() is not None

+    async def build_checkpoint(
+        self,
+        workflow_name: str,
+        graph_signature_hash: str,
+        state: State,
+        previous_checkpoint_id: CheckpointID | None,
+        iteration_count: int,
+        metadata: dict[str, Any] | None = None,
+    ) -> WorkflowCheckpoint:
+        return WorkflowCheckpoint(
+            workflow_name=workflow_name,
+            graph_signature_hash=graph_signature_hash,
+            previous_checkpoint_id=previous_checkpoint_id,
+            # Copy the per-source lists so the snapshot is isolated from later context mutations.
+            messages={source_id: list(messages) for source_id, messages in self._messages.items()},
+            state=state.export_state(),
+            pending_request_info_events=dict(self._pending_request_info_events),
+            iteration_count=iteration_count,
+            metadata=metadata or {},
+        )
+
    async def create_checkpoint(
        self,
        workflow_name: str,
@@ -394,15 +446,13 @@ class InProcRunnerContext:
        if not storage:
            raise ValueError("Checkpoint storage not configured")

-        checkpoint = WorkflowCheckpoint(
-            workflow_name=workflow_name,
-            graph_signature_hash=graph_signature_hash,
-            previous_checkpoint_id=previous_checkpoint_id,
-            messages=dict(self._messages),
-            state=state.export_state(),
-            pending_request_info_events=dict(self._pending_request_info_events),
-            iteration_count=iteration_count,
-            metadata=metadata or {},
+        checkpoint = await self.build_checkpoint(
+            workflow_name,
+            graph_signature_hash,
+            state,
+            previous_checkpoint_id,
+            iteration_count,
+            metadata,
        )
        checkpoint_id = await storage.save(checkpoint)
        logger.debug(f"Created checkpoint {checkpoint_id}")
@@ -814,10 +814,9 @@ class Workflow(DictConvertible):
            # runner context has fully drained from any prior run. If it still
            # has in-flight executor messages, the prior run didn't complete -
            # the caller must either resume from a checkpoint or wait for the
-            # prior run to drain. (Pending request_info events are intentionally
-            # NOT blocked here: a follow-up run with message=... is the normal
-            # way to deliver a response to those pending requests, e.g. via
-            # WorkflowAgent._process_pending_requests.)
+            # prior run to drain. Pending request_info events are intentionally
+            # NOT blocked here (they are answered via a follow-up ``responses=...``
+            # run); the warning below surfaces the abandon/overwrite cases instead.
            # NOTE: _validate_run_params already enforces that ``message`` is
            # mutually exclusive with both ``checkpoint_id`` and ``responses``,
            # so we don't need to re-check those here.
@@ -830,6 +829,33 @@ class Workflow(DictConvertible):
                    "checkpointing; there is no in-process recovery path."
                )

+            # Warn (but don't block) when a fresh message or a checkpoint restore begins while the
+            # workflow still has pending request_info events from an unfinished request/response
+            # cycle. A fresh ``message`` does NOT drop those pending requests - they remain pending and
+            # can still be answered later - but the new run advances executor and shared state, so when
+            # a response for an earlier request eventually arrives the workflow may have moved on,
+            # yielding inconsistent results. A ``checkpoint_id`` restore instead replaces the context's
+            # pending requests with the checkpoint's state. Delivering ``responses`` is the normal way to
+            # answer pending requests and is intentionally not warned. Mirrors the WorkflowExecutor
+            # warning for overlapping sub-workflow executions.
+            if message is not None or checkpoint_id is not None:
+                pending_request_info_events = await self._runner.context.get_pending_request_info_events()
+                if pending_request_info_events:
+                    logger.warning(
+                        "Workflow %s received %s while %d request_info event(s) are still pending from an "
+                        "unfinished request/response cycle; %s. Deliver responses (responses=...) to complete "
+                        "the pending cycle before starting new input.",
+                        self.id,
+                        "a fresh message" if message is not None else "a checkpoint restore",
+                        len(pending_request_info_events),
+                        (
+                            "those requests remain pending, but this run advances executor and shared state, "
+                            "so a response that arrives later may apply to a workflow that has moved on"
+                            if message is not None
+                            else "those pending requests will be overwritten by the checkpoint's state"
+                        ),
+                    )
+
            initial_executor_fn = self._resolve_execution_mode(message, responses, checkpoint_id, checkpoint_storage)

            async for event in self._run_workflow_with_tracing(
@@ -4,14 +4,12 @@ import asyncio
 import logging
 import sys
 import types
-import uuid
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any

 if TYPE_CHECKING:
    from ._workflow import Workflow

-from ._checkpoint_encoding import decode_checkpoint_value
 from ._const import GLOBAL_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY
 from ._events import (
    WorkflowEvent,
@@ -36,7 +34,12 @@ logger = logging.getLogger(__name__)

@dataclass
 class ExecutionContext:
-    """Context for tracking a single sub-workflow execution."""
+    """Legacy per-execution bookkeeping.
+
+    Retained only to decode checkpoints written before the sub-workflow's own checkpoint was
+    embedded (see ``WorkflowExecutor.on_checkpoint_restore``). It is no longer used at runtime -
+    the wrapped sub-workflow is the single source of truth for its pending requests.
+    """

    # The ID of the execution context
    execution_id: str
@@ -161,11 +164,9 @@ class WorkflowExecutor(Executor):
        # The response handler expects a SubWorkflowResponseMessage wrapping the response data.

    ### State Management
-    WorkflowExecutor maintains execution state across request/response cycles:
-    - Tracks pending requests by request_id
-    - Accumulates responses until all expected responses are received
-    - Resumes sub-workflow execution with complete response batch
-    - Handles concurrent executions and multiple pending requests
+    WorkflowExecutor keeps no request/response bookkeeping of its own. The wrapped sub-workflow
+    is the single source of truth for its pending requests; responses are forwarded to it and
+    validated against its own pending request_info events.

    ## Type System Integration
    WorkflowExecutor inherits its type signature from the wrapped workflow:
@@ -194,46 +195,21 @@ class WorkflowExecutor(Executor):
    - Converts to error event in parent context
    - Provides detailed error information including sub-workflow ID

-    ## Concurrent Execution Support
-    WorkflowExecutor fully supports multiple concurrent sub-workflow executions:
-
-    ### Per-Execution State Isolation
-    Each sub-workflow invocation creates an isolated ExecutionContext:
+    ## Overlapping Executions
+    A ``WorkflowExecutor`` wraps a single shared sub-workflow instance and keeps no per-execution
+    state. If a new input arrives while the sub-workflow still has pending request_info events from
+    an unfinished request/response cycle, the new input advances the shared sub-workflow state and
+    can interfere with that cycle - a response arriving later may apply to a sub-workflow that has
+    moved on. This is allowed but logs a warning, and is only safe when the wrapped workflow (and
+    its executors) are stateless.

    .. code-block:: python

-        # Multiple concurrent invocations are supported
-        workflow_executor = WorkflowExecutor(my_workflow, id="concurrent_executor")
-
-        # Each invocation gets its own execution context
-        # Execution 1: processes input_1 independently
-        # Execution 2: processes input_2 independently
-        # No state interference between executions
-
-    ### Request/Response Coordination
-    Responses are correctly routed to the originating execution:
-    - Each execution tracks its own pending requests and expected responses
-    - Request-to-execution mapping ensures responses reach the correct sub-workflow
-    - Response accumulation is isolated per execution
-    - Automatic cleanup when execution completes
-
-    ### Memory Management
-    - Unlimited concurrent executions supported
-    - Each execution has unique UUID-based identification
-    - Cleanup of completed execution contexts
-    - Thread-safe state management for concurrent access
-
-    ### Important Considerations
-    **Shared Workflow Instance**: All concurrent executions use the same underlying workflow instance.
-    For proper isolation, ensure that the wrapped workflow and its executors are stateless.
-
-    .. code-block:: python
-
-        # Avoid: Stateful executor with instance variables
+        # Avoid: stateful executor whose instance variables are shared across overlapping runs
        class StatefulExecutor(Executor):
            def __init__(self):
                super().__init__(id="stateful")
-                self.data = []  # This will be shared across concurrent executions!
+                self.data = []  # Shared across overlapping sub-workflow executions!

    ## Integration with Parent Workflows
    Parent workflows can intercept sub-workflow requests:
@@ -255,12 +231,18 @@ class WorkflowExecutor(Executor):
                    # Forward to external handler
                    await ctx.request_info(request.source_event, response_type=request.source_event.response_type)

+    ## Checkpointing
+    The provided sub workflow may not have its own checkpoint storage. The sub workflow checkpointed states will
+    be managed by the parent workflow.
+
    ## Implementation Notes
-    - Sub-workflows run to completion before processing their results
-    - Event processing is atomic - all outputs are forwarded before requests
-    - Response accumulation ensures sub-workflows receive complete response batches
-    - Execution state is maintained for proper resumption after external requests
-    - Concurrent executions are fully isolated and do not interfere with each other
+    - Sub-workflows run to completion (or to idle-with-pending-requests) before their results are processed
+    - Event processing is ordered - outputs are forwarded before requests
+    - Responses are forwarded to the sub-workflow as they arrive; the sub-workflow tracks its own
+      pending requests and resumes when they are answered
+    - The WorkflowExecutor keeps no per-execution bookkeeping; the sub-workflow is the single source
+      of truth for its pending requests. Starting a new execution while the sub-workflow still has
+      pending requests logs a warning and is only safe when the wrapped workflow is stateless
    """

    def __init__(
@@ -274,19 +256,18 @@ class WorkflowExecutor(Executor):
        """Initialize the WorkflowExecutor.

        Args:
-            workflow: The workflow to execute as a sub-workflow.
+            workflow: The workflow to execute as a sub-workflow. This workflow instance (including
+                the executor instances within it) must be unique. If the same instances are shared
+                across multiple WorkflowExecutor instances, it may lead to incorrect behavior.
            id: Unique identifier for this executor.
-            allow_direct_output: Whether to allow direct output from the sub-workflow.
-                                 By default, outputs from the sub-workflow are sent to
-                                 other executors in the parent workflow as messages.
-                                 When this is set to true, the outputs are yielded
-                                 directly from the WorkflowExecutor to the parent
-                                 workflow's event stream.
-            propagate_request: Whether to propagate requests from the sub-workflow to the
-                               parent workflow. If set to true, requests from the sub-workflow
-                               will be propagated as the original WorkflowEvent to the parent
-                               workflow. Otherwise, they will be wrapped in a SubWorkflowRequestMessage,
-                               which should be handled by an executor in the parent workflow.
+            allow_direct_output: Whether to allow direct output from the sub-workflow. By default,
+                outputs from the sub-workflow are sent to other executors in the parent workflow as
+                messages. When this is set to true, the outputs are yielded directly from the
+                WorkflowExecutor to the parent workflow's event stream.
+            propagate_request: Whether to propagate requests from the sub-workflow to the parent
+                workflow. If set to true, requests from the sub-workflow will be propagated as the
+                original WorkflowEvent to the parent workflow. Otherwise, they will be wrapped in a
+                SubWorkflowRequestMessage, which should be handled by an executor in the parent workflow.

        Keyword Args:
            **kwargs: Additional keyword arguments passed to the parent constructor.
@@ -294,13 +275,17 @@ class WorkflowExecutor(Executor):
        super().__init__(id, **kwargs)
        self.workflow = workflow
        self.allow_direct_output = allow_direct_output
-
-        # Track execution contexts for concurrent sub-workflow executions
-        self._execution_contexts: dict[str, ExecutionContext] = {}  # execution_id -> ExecutionContext
-        # Map request_id to execution_id for response routing
-        self._request_to_execution: dict[str, str] = {}  # request_id -> execution_id
        self._propagate_request = propagate_request

+        if self.workflow._runner_context.has_checkpointing():  # type: ignore
+            logger.warning(
+                "Sub workflow %s has its own checkpoint storage configured. "
+                "Sub workflow states are checkpointed by the parent workflow at superstep boundaries. "
+                "Additional checkpointing is only needed if you need to persist sub workflow state "
+                "independently of the parent workflow. ",
+                self.workflow.id,
+            )
+
    @property
    def input_types(self) -> list[type[Any] | types.UnionType]:
        """Get the input types based on the underlying workflow's input types plus WorkflowExecutor-specific types.
@@ -351,11 +336,10 @@ class WorkflowExecutor(Executor):
            # Always handle SubWorkflowResponseMessage
            return True

-        if (
-            message.original_request_info_event is not None
-            and message.original_request_info_event.request_id in self._request_to_execution
-        ):
-            # Handle propagated responses for known requests
+        if message.original_request_info_event is not None:
+            # A propagated response is target-routed back to the executor that issued the request,
+            # so if one reaches this WorkflowExecutor it belongs to our sub-workflow. _handle_response
+            # validates it against the sub-workflow's pending requests and ignores anything unknown.
            return True

        # For other messages, only handle if the wrapped workflow can accept them as input
@@ -372,58 +356,52 @@ class WorkflowExecutor(Executor):
            input_data: The input data to send to the sub-workflow.
            ctx: The workflow context from the parent.
        """
-        # Create execution context for this sub-workflow run
-        execution_id = str(uuid.uuid4())
-        execution_context = ExecutionContext(
-            execution_id=execution_id,
-            collected_responses={},
-            expected_response_count=0,
-            pending_requests={},
+        # The sub-workflow is a single shared instance. If it still has pending request_info events
+        # from an unfinished request/response cycle, a new input advances its shared state and can
+        # interfere with that cycle - a response arriving later may apply to a sub-workflow that has
+        # moved on. We allow it (the sub-workflow may be stateless) but warn so the risk is visible.
+        pending_requests = await self.workflow._runner_context.get_pending_request_info_events()  # pyright: ignore[reportPrivateUsage]
+        if pending_requests:
+            logger.warning(
+                f"WorkflowExecutor {self.id} received a new input message while its sub-workflow "
+                f"({self.workflow.id}) still has {len(pending_requests)} pending request(s) from an "
+                f"unfinished request/response cycle. The sub-workflow is a single shared instance, so the "
+                f"new input advances shared state and can interfere with the in-flight cycle. Ensure the "
+                f"sub-workflow is stateless, or complete the pending cycle before sending new input."
+            )
+
+        logger.debug(f"WorkflowExecutor {self.id} starting sub-workflow {self.workflow.id}")
+
+        # Get kwargs from parent workflow's State to propagate to subworkflow
+        parent_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {})
+
+        # Extract invocation kwargs recognised by Workflow.run()
+        # The state stores resolved format (with __global__ wrapper for global kwargs).
+        # Unwrap __global__ before passing to the subworkflow so it gets re-resolved
+        # against the subworkflow's own executor IDs.
+        fi_kwargs: dict[str, Any] | None = None
+        ci_kwargs: dict[str, Any] | None = None
+        for key in ("function_invocation_kwargs", "client_kwargs"):
+            resolved = parent_kwargs.get(key)
+            if isinstance(resolved, dict):
+                # Unwrap global sentinel; pass per-executor dicts as-is
+                unwrapped: dict[str, Any] = resolved.get(GLOBAL_KWARGS_KEY, resolved)  # type: ignore
+                if key == "function_invocation_kwargs":
+                    fi_kwargs = unwrapped  # type: ignore
+                else:
+                    ci_kwargs = unwrapped  # type: ignore
+
+        # Run the sub-workflow and collect all events, passing parent kwargs
+        result = await self.workflow.run(
+            input_data,
+            function_invocation_kwargs=fi_kwargs,  # type: ignore
+            client_kwargs=ci_kwargs,  # type: ignore
        )
-        self._execution_contexts[execution_id] = execution_context

-        logger.debug(f"WorkflowExecutor {self.id} starting sub-workflow {self.workflow.id} execution {execution_id}")
+        logger.debug(f"WorkflowExecutor {self.id} sub-workflow {self.workflow.id} completed with {len(result)} events")

-        try:
-            # Get kwargs from parent workflow's State to propagate to subworkflow
-            parent_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {})
-
-            # Extract invocation kwargs recognised by Workflow.run()
-            # The state stores resolved format (with __global__ wrapper for global kwargs).
-            # Unwrap __global__ before passing to the subworkflow so it gets re-resolved
-            # against the subworkflow's own executor IDs.
-            fi_kwargs: dict[str, Any] | None = None
-            ci_kwargs: dict[str, Any] | None = None
-            for key in ("function_invocation_kwargs", "client_kwargs"):
-                resolved = parent_kwargs.get(key)
-                if isinstance(resolved, dict):
-                    # Unwrap global sentinel; pass per-executor dicts as-is
-                    unwrapped: dict[str, Any] = resolved.get(GLOBAL_KWARGS_KEY, resolved)  # type: ignore
-                    if key == "function_invocation_kwargs":
-                        fi_kwargs = unwrapped  # type: ignore
-                    else:
-                        ci_kwargs = unwrapped  # type: ignore
-
-            # Run the sub-workflow and collect all events, passing parent kwargs
-            result = await self.workflow.run(
-                input_data,
-                function_invocation_kwargs=fi_kwargs,  # type: ignore
-                client_kwargs=ci_kwargs,  # type: ignore
-            )
-
-            logger.debug(
-                f"WorkflowExecutor {self.id} sub-workflow {self.workflow.id} "
-                f"execution {execution_id} completed with {len(result)} events"
-            )
-
-            # Process the workflow result using shared logic
-            await self._process_workflow_result(result, execution_context, ctx)
-        finally:
-            # Clean up execution context if it's completed (no pending requests)
-            if execution_id in self._execution_contexts:
-                exec_ctx = self._execution_contexts[execution_id]
-                if not exec_ctx.pending_requests:
-                    del self._execution_contexts[execution_id]
+        # Process the workflow result using shared logic
+        await self._process_workflow_result(result, ctx)

    @handler
    async def handle_message_wrapped_request_response(
@@ -433,8 +411,8 @@ class WorkflowExecutor(Executor):
    ) -> None:
        """Handle response from parent for a forwarded request.

-        This handler accumulates responses and only resumes the sub-workflow
-        when all expected responses have been received for that execution.
+        Forwards the response to the sub-workflow, which resumes and validates it against its
+        own pending requests.

        Args:
            response: The response to a previous request.
@@ -474,61 +452,44 @@ class WorkflowExecutor(Executor):
    async def on_checkpoint_save(self) -> dict[str, Any]:
        """Get the current state of the WorkflowExecutor for checkpointing purposes."""
        return {
-            "execution_contexts": {
-                execution_id: execution_context for execution_id, execution_context in self._execution_contexts.items()
-            },
-            "request_to_execution": dict(self._request_to_execution),
+            # The sub-workflow's own checkpoint carries everything needed to resume: shared state,
+            # executor snapshots, in-flight messages, and pending request_info events. The
+            # WorkflowExecutor keeps no separate request/response bookkeeping of its own. The
+            # sub-workflow is quiescent here: it ran to idle within this parent superstep before
+            # the parent checkpoints.
+            "sub_workflow_checkpoint": await self.workflow._runner.build_checkpoint(),  # pyright: ignore[reportPrivateUsage]
        }

    @override
    async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
        """Restore the WorkflowExecutor state from a checkpoint snapshot."""
-        # Validate the state contains the right keys
-        if "execution_contexts" not in state:
-            raise KeyError("Missing 'execution_contexts' in WorkflowExecutor state.")
-        if "request_to_execution" not in state:
-            raise KeyError("Missing 'request_to_execution' in WorkflowExecutor state.")
+        # The storage backend fully materializes the checkpoint on load, checkpointed data arrives as live objects.
+        sub_workflow_checkpoint = state.get("sub_workflow_checkpoint")
+        if sub_workflow_checkpoint is not None:
+            await self.workflow._runner.restore_checkpoint(sub_workflow_checkpoint)  # pyright: ignore[reportPrivateUsage]
+            return

-        # Validate the execution contexts stored in the state have the right keys and values
-        execution_contexts: dict[str, ExecutionContext] | None = None
-        try:
-            execution_contexts = {
-                key: decode_checkpoint_value(value) for key, value in state["execution_contexts"].items()
-            }
-        except Exception as ex:
-            raise RuntimeError("Failed to deserialize execution context.") from ex
-
-        if not all(
-            isinstance(key, str) and isinstance(value, ExecutionContext) for key, value in execution_contexts.items()
-        ):
-            raise ValueError("Execution contexts must have 'str' as key and 'ExecutionContext' as value.")
-        if not all(key == value.execution_id for key, value in execution_contexts.items()):
-            raise ValueError("Execution contexts must have matching keys and IDs.")
-
-        # Validate the request_to_execution map contain the right data
-        request_to_execution = state["request_to_execution"]
-        if not all(isinstance(key, str) and isinstance(value, str) for key, value in request_to_execution.items()):
-            raise ValueError("Request to execution map must have 'str' as key and 'str' as value.")
-        if not all(value in execution_contexts for value in request_to_execution.values()):
-            raise ValueError(
-                "'request_to_execution` contains unknown execution ID that is not part of the execution contexts."
-            )
-
-        self._execution_contexts = execution_contexts
-        self._request_to_execution = request_to_execution
-
-        # Add the `request_info_event`s back to the sub workflow.
-        # This is only a temporary solution to rehydrate the sub workflow with the requests.
-        # The proper way would be to rehydrate the workflow from a checkpoint on a Workflow
-        # API instead of the '_runner_context' object that should be hidden. And the sub workflow
-        # should be rehydrated from a checkpoint object instead of from a subset of the state.
-        # TODO(@taochen): Issue #1614 - how to handle the case when the parent workflow has checkpointing
-        # set up but not the sub workflow?
-        request_info_events = [
-            request_info_event
-            for execution_context in self._execution_contexts.values()
-            for request_info_event in execution_context.pending_requests.values()
-        ]
+        # Backward-compatibility fallback for checkpoints written before the sub-workflow checkpoint
+        # was embedded. Those stored per-execution bookkeeping; recover only the pending
+        # request_info events so the sub-workflow re-emits its pending requests. The sub-workflow's
+        # deeper executor/shared state cannot be restored from these older checkpoints.
+        legacy_execution_contexts = state.get("execution_contexts")
+        if not legacy_execution_contexts:
+            return
+        request_info_events: list[WorkflowEvent[Any]] = []
+        for execution_context in legacy_execution_contexts.values():
+            if isinstance(execution_context, ExecutionContext):
+                request_info_events.extend(execution_context.pending_requests.values())
+                if execution_context.collected_responses:
+                    logger.warning(
+                        "WorkflowExecutor %s restored legacy checkpoint with collected responses for "
+                        "execution_id %s. The sub-workflow is the single source of truth for its pending "
+                        "requests, so these responses will be ignored. Resume instead from a checkpoint created "
+                        "prior to any responses being collected if legacy request/response state must be "
+                        "preserved. Legacy execution contexts for sub-workflows will be removed in a future release.",
+                        self.id,
+                        execution_context.execution_id,
+                    )
        await asyncio.gather(*[
            self.workflow._runner_context.add_request_info_event(event)  # pyright: ignore[reportPrivateUsage]
            for event in request_info_events
@@ -537,7 +498,6 @@ class WorkflowExecutor(Executor):
    async def _process_workflow_result(
        self,
        result: WorkflowRunResult,
-        execution_context: ExecutionContext,
        ctx: WorkflowContext[Any],
    ) -> None:
        """Process the result from a workflow execution.
@@ -547,7 +507,6 @@ class WorkflowExecutor(Executor):

        Args:
            result: The workflow execution result.
-            execution_context: The execution context for this sub-workflow run.
            ctx: The workflow context.
        """
        # Collect all events from the workflow
@@ -586,10 +545,6 @@ class WorkflowExecutor(Executor):
        for event in request_info_events:
            request_id = event.request_id
            response_type = event.response_type
-            # Track the pending request in execution context
-            execution_context.pending_requests[request_id] = event
-            # Map request to execution for response routing
-            self._request_to_execution[request_id] = execution_context.execution_id
            if self._propagate_request:
                # In a workflow where the parent workflow does not handle the request, the request
                # should be propagated via the `request_info` mechanism to an external source. And
@@ -600,9 +555,6 @@ class WorkflowExecutor(Executor):
                # request and handle it directly, a message should be sent.
                await ctx.send_message(SubWorkflowRequestMessage(source_event=event, executor_id=self.id))

-        # Update expected response count for this execution
-        execution_context.expected_response_count = len(request_info_events)
-
        # Handle final state
        if workflow_run_state == WorkflowRunState.FAILED:
            # Find the failed event (type='failed').
@@ -621,26 +573,18 @@ class WorkflowExecutor(Executor):
                await ctx.add_event(error_event)
        elif workflow_run_state == WorkflowRunState.IDLE:
            # Sub-workflow is idle - nothing more to do now
-            logger.debug(
-                f"Sub-workflow {self.workflow.id} is idle with {len(self._execution_contexts)} active executions"
-            )
+            logger.debug(f"Sub-workflow {self.workflow.id} is idle")
        elif workflow_run_state == WorkflowRunState.CANCELLED:
            # Sub-workflow was cancelled - treat as completion
-            logger.debug(
-                f"Sub-workflow {self.workflow.id} was cancelled with {len(self._execution_contexts)} active executions"
-            )
+            logger.debug(f"Sub-workflow {self.workflow.id} was cancelled")
        elif workflow_run_state == WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS:
            # Sub-workflow is still running with pending requests
            logger.debug(
-                f"Sub-workflow {self.workflow.id} is still in progress with {len(request_info_events)} "
-                f"pending requests with {len(self._execution_contexts)} active executions"
+                f"Sub-workflow {self.workflow.id} is still in progress with {len(request_info_events)} pending requests"
            )
        elif workflow_run_state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS:
            # Sub-workflow is idle but has pending requests
-            logger.debug(
-                f"Sub-workflow {self.workflow.id} is idle with pending requests: "
-                f"{len(request_info_events)} with {len(self._execution_contexts)} active executions"
-            )
+            logger.debug(f"Sub-workflow {self.workflow.id} is idle with pending requests: {len(request_info_events)}")
        else:
            raise RuntimeError(f"Unexpected workflow run state: {workflow_run_state}")

@@ -650,48 +594,17 @@ class WorkflowExecutor(Executor):
        response: Any,
        ctx: WorkflowContext[Any],
    ) -> None:
-        execution_id = self._request_to_execution.get(request_id)
-        if not execution_id or execution_id not in self._execution_contexts:
+        # The sub-workflow is the source of truth for what it is awaiting. Validate the response
+        # against its pending requests and ignore anything unknown or already handled.
+        pending_requests = await self.workflow._runner_context.get_pending_request_info_events()  # pyright: ignore[reportPrivateUsage]
+        if request_id not in pending_requests:
            logger.warning(
-                f"WorkflowExecutor {self.id} received response for unknown request_id: {request_id}. "
-                "This response will be ignored."
+                f"WorkflowExecutor {self.id} received a response for an unknown or already-handled "
+                f"request_id: {request_id}. This response will be ignored."
            )
            return

-        execution_context = self._execution_contexts[execution_id]
-
-        # Check if we have this pending request in the execution context
-        if request_id not in execution_context.pending_requests:
-            logger.warning(
-                f"WorkflowExecutor {self.id} received response for unknown request_id: "
-                f"{request_id} in execution {execution_id}, ignoring"
-            )
-            return
-
-        # Remove the request from pending list and request mapping
-        execution_context.pending_requests.pop(request_id, None)
-        self._request_to_execution.pop(request_id, None)
-
-        # Accumulate the response in this execution's context
-        execution_context.collected_responses[request_id] = response
-        # Check if we have all expected responses for this execution
-        if len(execution_context.collected_responses) < execution_context.expected_response_count:
-            logger.debug(
-                f"WorkflowExecutor {self.id} execution {execution_id} waiting for more responses: "
-                f"{len(execution_context.collected_responses)}/{execution_context.expected_response_count} received"
-            )
-            return  # Wait for more responses
-
-        # Send all collected responses to the sub-workflow
-        responses_to_send = dict(execution_context.collected_responses)
-        execution_context.collected_responses.clear()  # Clear for next batch
-
-        try:
-            # Resume the sub-workflow with all collected responses
-            result = await self.workflow.run(responses=responses_to_send)
-            # Process the workflow result using shared logic
-            await self._process_workflow_result(result, execution_context, ctx)
-        finally:
-            # Clean up execution context if it's completed (no pending requests)
-            if not execution_context.pending_requests:
-                del self._execution_contexts[execution_id]
+        # Forward the response to the sub-workflow, which resumes and validates it against its own
+        # pending requests, then process whatever the sub-workflow produces.
+        result = await self.workflow.run(responses={request_id: response})
+        await self._process_workflow_result(result, ctx)
@@ -2338,6 +2338,42 @@ def test_replace_approval_contents_with_results_uses_result_call_ids_without_pla
    ]


+def test_replace_approval_contents_with_results_allows_reused_call_id_after_completion() -> None:
+    """A completed call must not suppress a later approval request that reuses its id.
+
+    Re-approving the same ``(call_id, function)`` is supported behaviour. If the dedupe
+    matched every occurrence of the id, the fresh request would be dropped and its result
+    attached to the already-answered call, leaving one call with two results.
+    """
+    from agent_framework._tools import _collect_approval_responses, _replace_approval_contents_with_results
+
+    completed_call = Content.from_function_call(call_id="call_reused", name="run_skill_script", arguments="{}")
+    completed_result = Content.from_function_result(call_id="call_reused", result="first output")
+    _, request, response = _build_approved_tool_roundtrip(
+        call_id="call_reused", approval_id="approval_2", tool_name="run_skill_script"
+    )
+
+    messages = [
+        Message(role="assistant", contents=[completed_call]),
+        Message(role="tool", contents=[completed_result]),
+        Message(role="assistant", contents=[request]),
+        Message(role="user", contents=[response]),
+    ]
+
+    _replace_approval_contents_with_results(
+        messages,
+        _collect_approval_responses(messages),
+        [Content.from_function_result(call_id="call_reused", result="second output")],
+    )
+
+    function_calls = [c for m in messages for c in m.contents if c.type == "function_call"]
+    assert [c.call_id for c in function_calls] == ["call_reused", "call_reused"]
+    results = [c for m in messages for c in m.contents if c.type == "function_result"]
+    assert [(c.call_id, c.result) for c in results] == [
+        ("call_reused", "first output"),
+        ("call_reused", "second output"),
+    ]
+
 def test_replace_approval_contents_with_results_uses_result_call_ids_for_placeholders() -> None:
    from agent_framework._tools import _collect_approval_responses, _replace_approval_contents_with_results

@@ -257,6 +257,43 @@ class TestHITL:
        assert outputs == ["Final: Looks great!"]
        assert result2.get_final_state() == WorkflowRunState.IDLE

+    async def test_fresh_message_while_pending_requests_warns(self, caplog: pytest.LogCaptureFixture) -> None:
+        """A fresh message while request_info events are pending is allowed but logs a warning."""
+
+        @workflow
+        async def review_wf(doc: str, ctx: RunContext) -> str:
+            feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1")
+            return f"Final: {feedback}"
+
+        result1 = await review_wf.run("my doc")
+        assert result1.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
+
+        # Starting fresh input while a request is pending does not abandon it, but advances
+        # workflow state so a later response may apply to a moved-on workflow -> warn (but proceed).
+        with caplog.at_level(logging.WARNING):
+            await review_wf.run("another doc")
+
+        assert "request_info event(s) are still pending" in caplog.text
+        assert "a fresh message" in caplog.text
+
+    async def test_responses_while_pending_requests_does_not_warn(self, caplog: pytest.LogCaptureFixture) -> None:
+        """Delivering responses is the normal completion path and must not warn."""
+
+        @workflow
+        async def review_wf(doc: str, ctx: RunContext) -> str:
+            feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1")
+            return f"Final: {feedback}"
+
+        result1 = await review_wf.run("my doc")
+        assert result1.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
+
+        caplog.clear()
+        with caplog.at_level(logging.WARNING):
+            result2 = await review_wf.run(responses={"req1": "Looks great!"})
+
+        assert result2.get_final_state() == WorkflowRunState.IDLE
+        assert "still pending" not in caplog.text
+
    async def test_untyped_ctx_parameter(self):
        """ctx is injected by parameter name even without a RunContext annotation."""

@@ -512,6 +512,98 @@ async def test_runner_reset_iteration_count():
    assert runner._iteration == 0  # pyright: ignore[reportPrivateUsage]


+async def test_runner_capture_and_restore_checkpoint_object_roundtrip():
+    """build_checkpoint() then restore_checkpoint() must roundtrip.
+
+    Shared state and executor snapshots are captured into an in-memory ``WorkflowCheckpoint``
+    and restored from it without any storage backend (the path the parent WorkflowExecutor
+    uses to checkpoint a nested sub-workflow).
+    """
+
+    class CounterExecutor(Executor):
+        def __init__(self, id: str) -> None:
+            super().__init__(id=id)
+            self.count = 0
+
+        @handler
+        async def handle(self, message: MockMessage, ctx: WorkflowContext[Any, int]) -> None:
+            self.count += message.data
+
+        async def on_checkpoint_save(self) -> dict[str, Any]:
+            return {"count": self.count}
+
+        async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
+            self.count = int(state.get("count", 0))
+
+    executor = CounterExecutor(id="counter")
+    state = State()
+    ctx = InProcRunnerContext()
+    runner = Runner([], {executor.id: executor}, state, ctx, "test_name", graph_signature_hash="test_hash")
+
+    # Establish some state to capture.
+    executor.count = 7
+    state.set("shared_key", "shared_value")
+    state.commit()
+
+    checkpoint = await runner.build_checkpoint()
+    assert checkpoint.graph_signature_hash == "test_hash"
+
+    # Mutate after capture; restoring must roll back to the captured snapshot.
+    executor.count = 999
+    state.set("shared_key", "mutated")
+    state.commit()
+
+    await runner.restore_checkpoint(checkpoint)
+
+    assert executor.count == 7
+    assert state.get("shared_key") == "shared_value"
+    assert runner._resumed_from_checkpoint is True  # pyright: ignore[reportPrivateUsage]
+
+
+async def test_runner_build_checkpoint_includes_in_flight_messages():
+    """build_checkpoint() must snapshot in-flight messages non-destructively."""
+    executor = MockExecutor(id="executor_a")
+    state = State()
+    ctx = InProcRunnerContext()
+    runner = Runner([], {executor.id: executor}, state, ctx, "test_name", graph_signature_hash="test_hash")
+
+    await ctx.send_message(WorkflowMessage(data=MockMessage(data=1), source_id="START"))
+
+    checkpoint = await runner.build_checkpoint()
+
+    # The in-flight message is captured in the snapshot ...
+    assert list(checkpoint.messages.keys()) == ["START"]
+    assert len(checkpoint.messages["START"]) == 1
+    # ... without draining it from the runner (capture is non-destructive).
+    assert await ctx.has_messages() is True
+
+
+async def test_runner_build_checkpoint_do_not_advance_previous_checkpoint_id():
+    """build_checkpoint() must not advance _previous_checkpoint_id so a later capture chains to it."""
+    executor = MockExecutor(id="executor_a")
+    state = State()
+    ctx = InProcRunnerContext()
+    runner = Runner([], {executor.id: executor}, state, ctx, "test_name", graph_signature_hash="test_hash")
+
+    # Pre-condition: nothing captured yet, so there is no parent to chain back to.
+    assert runner._previous_checkpoint_id is None  # pyright: ignore[reportPrivateUsage]
+
+    first = await runner.build_checkpoint()
+    assert first.previous_checkpoint_id is None
+
+    # Capturing advances the tracked checkpoint id to the newly-created checkpoint ...
+    assert runner._previous_checkpoint_id is None  # pyright: ignore[reportPrivateUsage]
+
+
+async def test_runner_restore_checkpoint_rejects_graph_mismatch():
+    """restore_checkpoint() must reject a checkpoint from a different graph."""
+    runner = Runner([], {}, State(), InProcRunnerContext(), "test_name", graph_signature_hash="hash-a")
+
+    foreign = WorkflowCheckpoint(workflow_name="test_name", graph_signature_hash="hash-b")
+    with pytest.raises(WorkflowCheckpointException, match="Workflow graph has changed"):
+        await runner.restore_checkpoint(foreign)
+
+
 class CheckpointingContext(InProcRunnerContext):
    """A context that supports checkpointing for testing."""

@@ -1,9 +1,11 @@
 # Copyright (c) Microsoft. All rights reserved.

+import logging
 from dataclasses import dataclass, field
 from typing import Any
 from uuid import uuid4

+import pytest
 from typing_extensions import Never

 from agent_framework import (
@@ -15,6 +17,7 @@ from agent_framework import (
    WorkflowContext,
    WorkflowEvent,
    WorkflowExecutor,
+    WorkflowRunState,
    handler,
    response_handler,
 )
@@ -465,6 +468,62 @@ async def test_concurrent_sub_workflow_execution() -> None:
    # (This is implicitly tested by the fact that we got correct results for all emails)


+async def test_sub_workflow_warns_on_overlapping_execution(caplog: pytest.LogCaptureFixture) -> None:
+    """A new input while a prior sub-workflow execution is awaiting responses logs a warning.
+
+    Overlapping executions share one sub-workflow instance and its state, so WorkflowExecutor
+    allows the new execution but warns that it is only safe when the wrapped workflow is stateless.
+    """
+
+    class TwoInputParent(Executor):
+        def __init__(self) -> None:
+            super().__init__(id="two_input_parent")
+            self._pending: dict[str, SubWorkflowRequestMessage] = {}
+
+        @handler
+        async def start(self, emails: list[str], ctx: WorkflowContext[EmailValidationRequest]) -> None:
+            for email in emails:
+                await ctx.send_message(EmailValidationRequest(email=email))
+
+        @handler
+        async def handle_domain_request(
+            self,
+            sub_workflow_request: SubWorkflowRequestMessage,
+            ctx: WorkflowContext[SubWorkflowResponseMessage],
+        ) -> None:
+            domain_request = sub_workflow_request.source_event.data
+            assert isinstance(domain_request, DomainCheckRequest)
+            self._pending[domain_request.id] = sub_workflow_request
+            await ctx.request_info(domain_request, bool)
+
+        @handler
+        async def collect(self, result: ValidationResult, ctx: WorkflowContext) -> None: ...
+
+    parent = TwoInputParent()
+    workflow_executor = WorkflowExecutor(create_email_validation_workflow(), "email_workflow")
+    main_workflow = (
+        WorkflowBuilder(start_executor=parent)
+        .add_edge(parent, workflow_executor)
+        .add_edge(workflow_executor, parent)
+        .build()
+    )
+
+    with caplog.at_level(logging.WARNING, logger="agent_framework._workflows._workflow_executor"):
+        result = await main_workflow.run(["a@domain1.com", "b@domain2.com"])
+
+    # Two inputs are delivered to the same WorkflowExecutor in one superstep: the second execution
+    # starts while the sub-workflow still has a pending request, producing exactly one overlap
+    # warning from the WorkflowExecutor. (The substring is unique to the WorkflowExecutor warning so
+    # it is not confused with the core Workflow.run pending-request warning.)
+    assert len(result.get_request_info_events()) == 2
+    overlap_warnings = [
+        record
+        for record in caplog.records
+        if record.levelno == logging.WARNING and "new input message while its sub-workflow" in record.getMessage()
+    ]
+    assert len(overlap_warnings) == 1
+
+
 # region Checkpoint-related message types and executors for sub-workflow tests


@@ -619,6 +678,59 @@ async def test_sub_workflow_checkpoint_restore_no_duplicate_requests() -> None:
    assert request_events[0].data.prompt == "Second request"


+async def test_sub_workflow_checkpoint_restore_preserves_sub_workflow_state() -> None:
+    """Resuming a sub-workflow mid-progress must restore its internal executor state.
+
+    Regression guard for the issue where only the WorkflowExecutor's bookkeeping (pending
+    requests) was checkpointed, so a sub-workflow executor that accumulates state across
+    multiple request/response cycles (here ``TwoStepSubWorkflowExecutor._responses``) lost
+    that state on resume. With the sub-workflow's own checkpoint embedded in the parent
+    checkpoint, the second response now completes the two-step flow instead of triggering a
+    spurious third request.
+    """
+    storage = InMemoryCheckpointStorage()
+
+    # Step 1: run until the first request.
+    workflow1 = _build_checkpoint_test_workflow(storage)
+    first_request_id: str | None = None
+    async for event in workflow1.run("test_value", stream=True):
+        if event.type == "request_info":
+            first_request_id = event.request_id
+    assert first_request_id is not None
+
+    # Step 2: answer the first request so the sub-workflow accumulates internal state
+    # (``_responses == ["first_answer"]``) and emits the second request. This mid-progress
+    # point is what we checkpoint and resume from - the case the no-duplicate test (which
+    # checkpoints at the first request, before any state accrues) does not cover.
+    second_request_id: str | None = None
+    async for event in workflow1.run(stream=True, responses={first_request_id: "first_answer"}):
+        if event.type == "request_info":
+            second_request_id = event.request_id
+    assert second_request_id is not None
+
+    # Resume from the latest checkpoint (captured after the second request was made).
+    checkpoints = await storage.list_checkpoints(workflow_name=workflow1.name)
+    checkpoint_id = max(checkpoints, key=lambda cp: cp.iteration_count).checkpoint_id
+
+    workflow2 = _build_checkpoint_test_workflow(storage)
+    resumed_second_request_id: str | None = None
+    async for event in workflow2.run(checkpoint_id=checkpoint_id, stream=True):
+        if event.type == "request_info":
+            resumed_second_request_id = event.request_id
+    assert resumed_second_request_id is not None
+    assert resumed_second_request_id == second_request_id
+
+    # Step 3: answer the second request. With the sub-workflow's state restored, the two-step
+    # executor completes instead of emitting a spurious third request. If the internal state
+    # were lost, the second answer would be treated as a first answer and a third request
+    # ("Second request") would be emitted.
+    result = await workflow2.run(responses={resumed_second_request_id: "second_answer"})
+    assert result.get_request_info_events() == [], (
+        "Sub-workflow internal state was lost on resume: a spurious extra request was emitted"
+    )
+    assert result.get_final_state() == WorkflowRunState.IDLE
+
+
 async def test_sub_workflow_intermediate_outputs_propagate_to_parent() -> None:
    """A child workflow's intermediate emissions must bubble up through the parent.

@@ -2,6 +2,7 @@

 import asyncio
 import gc
+import logging
 import tempfile
 from collections.abc import AsyncIterable, Awaitable, Sequence
 from dataclasses import dataclass, field
@@ -109,6 +110,38 @@ class MockExecutorRequestApproval(Executor):
            await ctx.send_message(NumberMessage(data=data))


+async def test_fresh_message_while_pending_advances_state_without_abandoning_requests(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A fresh message while a request is pending is allowed but hazardous.
+
+    A fresh ``message`` does NOT abandon the pending request - it can still be answered
+    later - but the new run advances executor state, so a response for the earlier request
+    applies to a workflow that has moved on. The run is allowed and a warning is emitted.
+    """
+    executor = MockExecutorRequestApproval(id="approver")
+    workflow = WorkflowBuilder(start_executor=executor).build()
+
+    # Turn 1: request approval for data=1 -> workflow idles with a pending request.
+    result1 = await workflow.run(NumberMessage(data=1))
+    assert result1.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
+    original_request_id = result1.get_request_info_events()[0].request_id
+
+    # Turn 2: a fresh message for data=2 while the first request is still pending. This is
+    # allowed but warns, and advances the executor's stored state from 1 to 2.
+    with caplog.at_level(logging.WARNING):
+        result2 = await workflow.run(NumberMessage(data=2))
+    assert "request_info event(s) are still pending" in caplog.text
+    assert "a fresh message" in caplog.text
+    assert result2.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
+
+    # Turn 3: the ORIGINAL request is still answerable, proving the fresh message did not
+    # abandon it. But because the executor state moved on to 2, the response applies to the
+    # moved-on state and yields 2, not the original 1.
+    result3 = await workflow.run(responses={original_request_id: ApprovalMessage(approved=True)})
+    assert result3.get_outputs() == [2]
+
+
 async def test_workflow_run_streaming() -> None:
    """Test the workflow run stream."""
    executor_a = IncrementExecutor(id="executor_a")
@@ -109,6 +109,17 @@ class CapturingRunnerContext(RunnerContext):
    ) -> str:
        raise NotImplementedError("Checkpointing is not supported in activity context")

+    async def build_checkpoint(
+        self,
+        workflow_name: str,
+        graph_signature_hash: str,
+        state: State,
+        previous_checkpoint_id: str | None,
+        iteration_count: int,
+        metadata: dict[str, Any] | None = None,
+    ) -> WorkflowCheckpoint:
+        raise NotImplementedError("Checkpointing is not supported in activity context")
+
    async def load_checkpoint(self, checkpoint_id: str) -> WorkflowCheckpoint | None:
        raise NotImplementedError("Checkpointing is not supported in activity context")

@@ -395,10 +395,32 @@ args = [
 ]

 [tool.poe.tasks.validate-dependency-bounds-test]
-help = "Run workspace dependency-bound validation in test mode, optionally scoped with -P/--package short names such as `core`."
+help = "Run the exhaustive workspace dependency-bound test+typing matrix, optionally scoped with -P/--package short names such as `core`."
 shell = "python -m scripts.dependencies.validate_dependency_bounds --mode test --package \"$project\""
 args = [{ name = "project", default = "*", options = ["-P", "--package"] }]

+[tool.poe.tasks.validate-python-release]
+help = "Refresh uv.lock, then run lower/upper import probes for changed package metadata on each package closure's minimum Python."
+executor = "simple"
+shell = """
+command=(
+  python -m scripts.dependencies.validate_dependency_bounds
+  --mode release
+  --base-ref "${base_ref}"
+  --release-timeout-seconds "${timeout}"
+)
+if [ -n "${python}" ]; then
+  command+=(--python "${python}")
+fi
+"${command[@]}"
+"""
+interpreter = "bash"
+args = [
+    { name = "base_ref", options = ["-B", "--base-ref"] },
+    { name = "python", default = "", options = ["--python"] },
+    { name = "timeout", default = "300", options = ["--timeout-seconds"] },
+]
+
 [tool.poe.tasks.validate-dependency-bounds-project]
 help = "Validate lower and upper dependency bounds for a -P/--package workspace package, optionally narrowed with -M/--mode and -D/--dependency."
 shell = """
@@ -12,10 +12,19 @@ Run the commands below from the `python/` directory.

 - `validate_dependency_bounds.py`
  - Main entrypoint for dependency-bound workflows.
-  - Supports `test`, `lower`, `upper`, and `both` modes.
-  - `test` runs workspace-wide smoke validation at the lower and upper ends of the currently allowed ranges.
+  - Supports `release`, `test`, `lower`, `upper`, and `both` modes.
+  - `release` refreshes `uv.lock`, then runs changed packages through fast lock-independent lower/upper import probes.
+  - `test` runs the exhaustive workspace test+typing compatibility matrix.
  - `lower`, `upper`, and `both` dispatch to the lower/upper optimizer implementations for one package.

+- `_dependency_bounds_release_impl.py`
+  - Discovers package metadata changed from the selected release base.
+  - Resolves published runtime dependencies and non-development extras independently of `uv.lock` with both
+    `lowest-direct` and `highest` strategies.
+  - Derives the minimum supported Python minor from each changed package's internal editable dependency closure.
+  - Imports each changed package and records resolved dependency versions in a JSON report.
+  - Runs probes concurrently under one five-minute deadline.
+
 - `upgrade_dev_dependencies.py`
  - Refreshes exact dev dependency pins across the root `pyproject.toml` and package `pyproject.toml` files.
  - Reuses the same version-selection logic as the upper-bound tooling so direct dev-tooling refreshes and dependency-range expansion stay consistent.
@@ -45,6 +54,7 @@ These are the normal user-facing entrypoints:
 ```bash
 uv run poe upgrade-dev-dependency-pins
 uv run poe upgrade-dev-dependencies
+uv run poe validate-python-release --base-ref upstream/main
 uv run poe validate-dependency-bounds-test
 uv run poe validate-dependency-bounds-test --package core
 uv run poe validate-dependency-bounds-project --mode both --package core --dependency "<dependency-name>"
@@ -52,7 +62,10 @@ uv run poe validate-dependency-bounds-project --mode both --package core --depen

 - `upgrade-dev-dependency-pins` only refreshes exact dev pins in `pyproject.toml` files.
 - `upgrade-dev-dependencies` refreshes dev pins (using task above), runs `uv lock --upgrade`, reinstalls from the frozen lockfile, then runs `check`, `typing`, and `test`.
- `validate-dependency-bounds-test` runs the repo-wide lower/upper smoke gate.
+- `validate-python-release` is the bounded release gate: it refreshes `uv.lock`, finds changed package metadata,
+  and probes both dependency-bound extremes without reusing the lockfile.
+- `validate-dependency-bounds-test` runs the exhaustive package test+typing matrix and is intentionally not part of
+  the routine release path.
 - `validate-dependency-bounds-project` is the single package-scoped task; use `--mode lower`, `--mode upper`, or `--mode both` for the target package/dependency pair. Its `--package` argument defaults to `*`, and `--dependency` is optional, so automation can also use it for repo-wide upper-bound runs.

 ### GitHub Actions workflows
@@ -76,6 +89,7 @@ These are useful for debugging or targeted manual runs:

 ```bash
 python -m scripts.dependencies.upgrade_dev_dependencies --dry-run --version-source lock
+python -m scripts.dependencies.validate_dependency_bounds --mode release --base-ref upstream/main --dry-run
 python -m scripts.dependencies.validate_dependency_bounds --mode test --package core --dry-run
 python -m scripts.dependencies.validate_dependency_bounds --mode both --package core --dependencies openai --dry-run
 python -m scripts.dependencies._dependency_bounds_lower_impl --packages core --dependencies openai --dry-run
@@ -89,6 +103,7 @@ Use the direct lower/upper implementation modules mainly for debugging or develo
 The validators write JSON reports into this folder:

 - `dependency-bounds-test-results.json`
+- `dependency-bounds-release-results.json`
 - `dependency-lower-bound-results.json`
 - `dependency-range-results.json`

@@ -0,0 +1,574 @@
+# Copyright (c) Microsoft. All rights reserved.
+# ruff:file-ignore[suspicious-subprocess-import, subprocess-without-shell-equals-true]
+
+"""Fast, lock-independent dependency-bound probes for Python release cuts."""
+
+from __future__ import annotations
+
+import concurrent.futures
+import json
+import os
+import subprocess
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, cast
+
+import tomli
+from packaging.requirements import InvalidRequirement, Requirement
+from packaging.specifiers import SpecifierSet
+from packaging.utils import canonicalize_name
+from packaging.version import Version
+from rich import print
+
+from scripts.task_runner import discover_projects, project_filter_matches
+
+_PROBE_RESULT_PREFIX = "DEPENDENCY_BOUNDS_RELEASE_RESULT="
+_RESOLUTION_SCENARIOS = (("lower", "lowest-direct"), ("upper", "highest"))
+
+
+@dataclass
+class ReleaseProject:
+    """Published metadata needed to build a release probe."""
+
+    project_path: Path
+    package_name: str
+    requires_python: str
+    dependencies: tuple[str, ...]
+    optional_dependencies: dict[str, tuple[str, ...]]
+    import_modules: tuple[str, ...]
+
+
+@dataclass
+class ReleaseProbePlan:
+    """One changed package and the local projects needed to resolve it."""
+
+    project_path: Path
+    package_name: str
+    editable_specs: tuple[str, ...]
+    import_modules: tuple[str, ...]
+    reported_distributions: tuple[str, ...]
+    python_version: str
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _write_json(path: Path, payload: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, sort_keys=False))
+
+
+def _truncate_error(stdout: str, stderr: str, *, max_chars: int = 3000) -> str:
+    combined = "\n".join(part for part in (stderr.strip(), stdout.strip()) if part)
+    if len(combined) <= max_chars:
+        return combined
+    return f"...\n{combined[-max_chars:]}"
+
+
+def _string_requirements(values: object) -> tuple[str, ...]:
+    if not isinstance(values, list):
+        return ()
+    return tuple(value for value in cast(list[object], values) if isinstance(value, str))
+
+
+def _discover_import_modules(project_path: Path, config: dict[str, Any]) -> tuple[str, ...]:
+    """Discover top-level import names from the project's build configuration."""
+    modules: set[str] = set()
+    tool = cast(dict[str, Any], config.get("tool", {}) or {})
+
+    flit = cast(dict[str, Any], tool.get("flit", {}) or {})
+    flit_module_config = cast(dict[str, Any], flit.get("module", {}) or {})
+    flit_module = flit_module_config.get("name")
+    if isinstance(flit_module, str) and flit_module:
+        modules.add(flit_module)
+
+    hatch = cast(dict[str, Any], tool.get("hatch", {}) or {})
+    hatch_build = cast(dict[str, Any], hatch.get("build", {}) or {})
+    hatch_targets = cast(dict[str, Any], hatch_build.get("targets", {}) or {})
+    hatch_wheel = cast(dict[str, Any], hatch_targets.get("wheel", {}) or {})
+    hatch_packages = hatch_wheel.get("packages", [])
+    if isinstance(hatch_packages, list):
+        for package in cast(list[object], hatch_packages):
+            if isinstance(package, str) and package:
+                modules.add(Path(package).name.split(".", 1)[0])
+
+    setuptools = cast(dict[str, Any], tool.get("setuptools", {}) or {})
+    setuptools_packages = setuptools.get("packages", [])
+    if isinstance(setuptools_packages, list):
+        for package in cast(list[object], setuptools_packages):
+            if isinstance(package, str) and package:
+                modules.add(package.split(".", 1)[0])
+
+    if not modules:
+        for candidate in project_path.glob("agent_framework*"):
+            if candidate.is_dir() and (candidate / "__init__.py").exists():
+                modules.add(candidate.name)
+            elif candidate.is_file() and candidate.suffix == ".py":
+                modules.add(candidate.stem)
+
+    return tuple(sorted(modules))
+
+
+def _load_release_project(workspace_root: Path, project_path: Path) -> ReleaseProject:
+    pyproject_file = workspace_root / project_path / "pyproject.toml"
+    with pyproject_file.open("rb") as file:
+        config = tomli.load(file)
+
+    project = cast(dict[str, Any], config.get("project", {}) or {})
+    package_name = str(project.get("name", "")).strip()
+    if not package_name:
+        raise RuntimeError(f"Missing project.name in {pyproject_file}")
+    requires_python = str(project.get("requires-python", "")).strip()
+    if not requires_python:
+        raise RuntimeError(f"Missing project.requires-python in {pyproject_file}")
+
+    optional_dependencies: dict[str, tuple[str, ...]] = {}
+    optional_config = cast(dict[str, object], project.get("optional-dependencies", {}) or {})
+    for extra_name, requirements in optional_config.items():
+        optional_dependencies[extra_name] = _string_requirements(requirements)
+
+    return ReleaseProject(
+        project_path=project_path,
+        package_name=package_name,
+        requires_python=requires_python,
+        dependencies=_string_requirements(project.get("dependencies", [])),
+        optional_dependencies=optional_dependencies,
+        import_modules=_discover_import_modules(pyproject_file.parent, config),
+    )
+
+
+def _build_release_project_map(workspace_root: Path) -> dict[str, ReleaseProject]:
+    project_paths = [Path("."), *sorted(set(discover_projects(workspace_root / "pyproject.toml")))]
+    projects: dict[str, ReleaseProject] = {}
+    for project_path in project_paths:
+        pyproject_file = workspace_root / project_path / "pyproject.toml"
+        if not pyproject_file.exists():
+            continue
+        project = _load_release_project(workspace_root, project_path)
+        projects[canonicalize_name(project.package_name)] = project
+    return projects
+
+
+def _changed_release_project_paths(workspace_root: Path, base_ref: str) -> set[Path]:
+    command = [
+        "git",
+        "diff",
+        "--relative",
+        "--name-only",
+        "--diff-filter=ACMR",
+        base_ref,
+        "--",
+        "pyproject.toml",
+        "packages/*/pyproject.toml",
+    ]
+    result = subprocess.run(command, cwd=workspace_root, capture_output=True, text=True, check=False)
+    if result.returncode != 0:
+        error = _truncate_error(result.stdout, result.stderr)
+        raise RuntimeError(f"Unable to compare release metadata with {base_ref}.\n{error}")
+
+    project_paths: set[Path] = set()
+    for line in result.stdout.splitlines():
+        changed_file = Path(line.strip())
+        if changed_file == Path("pyproject.toml"):
+            project_paths.add(Path("."))
+        elif len(changed_file.parts) == 3 and changed_file.parts[0] == "packages":
+            project_paths.add(changed_file.parent)
+    return project_paths
+
+
+def _selected_release_projects(
+    *,
+    workspace_root: Path,
+    projects: dict[str, ReleaseProject],
+    base_ref: str,
+    package_filter: str | None,
+) -> list[ReleaseProject]:
+    if package_filter:
+        selected = [
+            project
+            for project in projects.values()
+            if project_filter_matches(project.project_path, package_filter, [project.package_name])
+        ]
+    else:
+        changed_paths = _changed_release_project_paths(workspace_root, base_ref)
+        selected = [project for project in projects.values() if project.project_path in changed_paths]
+
+    return sorted(selected, key=lambda project: str(project.project_path))
+
+
+def _requirements_for_extras(project: ReleaseProject, extras: set[str]) -> tuple[str, ...]:
+    requirements = list(project.dependencies)
+    for extra_name in sorted(extras):
+        requirements.extend(project.optional_dependencies.get(extra_name, ()))
+    return tuple(requirements)
+
+
+def _minimum_python_version(projects: list[ReleaseProject]) -> str:
+    """Return the lowest Python minor supported by every project in a probe closure."""
+    constraints = [project.requires_python for project in projects]
+    combined = SpecifierSet(",".join(constraints))
+    lower_bounds = [
+        Version(specifier.version.rstrip(".*"))
+        for specifier in combined
+        if specifier.operator in {">", ">=", "~=", "=="} and specifier.version.rstrip(".*")
+    ]
+    if not lower_bounds:
+        package_names = ", ".join(sorted(project.package_name for project in projects))
+        raise RuntimeError(f"Unable to derive a Python floor from requires-python for: {package_names}")
+
+    floor = max(lower_bounds)
+    python_version = f"{floor.major}.{floor.minor}"
+    first_patch = Version(python_version)
+    later_patch = Version(f"{python_version}.999999")
+    if first_patch not in combined and later_patch not in combined:
+        package_names = ", ".join(sorted(project.package_name for project in projects))
+        raise RuntimeError(
+            f"No Python {python_version} interpreter satisfies the combined requires-python constraints for: "
+            f"{package_names}"
+        )
+    return python_version
+
+
+def _build_release_probe_plan(
+    workspace_root: Path,
+    target: ReleaseProject,
+    projects: dict[str, ReleaseProject],
+) -> ReleaseProbePlan:
+    """Build the exact internal editable closure for one changed package."""
+    target_name = canonicalize_name(target.package_name)
+    # Development extras are contributor tooling, not runtime compatibility surface.
+    requested_extras: dict[str, set[str]] = {
+        target_name: {extra for extra in target.optional_dependencies if extra != "dev"}
+    }
+    processed_extras: dict[str, set[str]] = {}
+    pending = [target_name]
+
+    while pending:
+        package_name = pending.pop()
+        project = projects[package_name]
+        extras = requested_extras[package_name]
+        if processed_extras.get(package_name) == extras:
+            continue
+        processed_extras[package_name] = set(extras)
+
+        for requirement_text in _requirements_for_extras(project, extras):
+            try:
+                requirement = Requirement(requirement_text)
+            except InvalidRequirement:
+                continue
+            dependency_name = canonicalize_name(requirement.name)
+            if dependency_name not in projects:
+                continue
+            previous = requested_extras.setdefault(dependency_name, set())
+            updated = previous | set(requirement.extras)
+            if dependency_name not in processed_extras or updated != previous:
+                requested_extras[dependency_name] = updated
+                pending.append(dependency_name)
+
+    target_extras = sorted(requested_extras[target_name])
+    target_path = (workspace_root / target.project_path).resolve()
+    target_spec = str(target_path)
+    if target_extras:
+        target_spec = f"{target_spec}[{','.join(target_extras)}]"
+
+    editable_specs = [target_spec]
+    for package_name in sorted(requested_extras):
+        if package_name == target_name:
+            continue
+        editable_specs.append(str((workspace_root / projects[package_name].project_path).resolve()))
+
+    target_requirements = _requirements_for_extras(target, set(target_extras))
+    reported_distributions = {canonicalize_name(target.package_name)}
+    for requirement_text in target_requirements:
+        try:
+            reported_distributions.add(canonicalize_name(Requirement(requirement_text).name))
+        except InvalidRequirement:
+            continue
+
+    return ReleaseProbePlan(
+        project_path=target.project_path,
+        package_name=target.package_name,
+        editable_specs=tuple(editable_specs),
+        import_modules=target.import_modules,
+        reported_distributions=tuple(sorted(reported_distributions)),
+        python_version=_minimum_python_version([projects[package_name] for package_name in requested_extras]),
+    )
+
+
+def _build_release_probe_command(
+    plan: ReleaseProbePlan,
+    *,
+    resolution: str,
+    python_override: str | None = None,
+) -> list[str]:
+    probe_script = f"""
+import importlib
+import json
+from importlib.metadata import PackageNotFoundError, version
+
+modules = {plan.import_modules!r}
+distributions = {plan.reported_distributions!r}
+for module_name in modules:
+    importlib.import_module(module_name)
+versions = {{}}
+for distribution_name in distributions:
+    try:
+        versions[distribution_name] = version(distribution_name)
+    except PackageNotFoundError:
+        versions[distribution_name] = None
+print({_PROBE_RESULT_PREFIX!r} + json.dumps({{"imports": modules, "versions": versions}}, sort_keys=True))
+"""
+    command = [
+        "uv",
+        "--no-progress",
+        "run",
+        "--isolated",
+        "--no-project",
+        "--python",
+        python_override or plan.python_version,
+        "--resolution",
+        resolution,
+        "--prerelease",
+        "if-necessary-or-explicit",
+        "--quiet",
+    ]
+    for editable_spec in plan.editable_specs:
+        command.extend(["--with-editable", editable_spec])
+    command.extend(["python", "-c", probe_script])
+    return command
+
+
+def _parse_probe_payload(stdout: str) -> dict[str, Any] | None:
+    for line in reversed(stdout.splitlines()):
+        if line.startswith(_PROBE_RESULT_PREFIX):
+            try:
+                payload = json.loads(line.removeprefix(_PROBE_RESULT_PREFIX))
+            except json.JSONDecodeError:
+                return None
+            return cast(dict[str, Any], payload) if isinstance(payload, dict) else None
+    return None
+
+
+def _run_release_probe(
+    plan: ReleaseProbePlan,
+    *,
+    scenario_name: str,
+    resolution: str,
+    python_override: str | None,
+    deadline: float,
+    dry_run: bool,
+) -> dict[str, Any]:
+    python_version = python_override or plan.python_version
+    command = _build_release_probe_command(plan, resolution=resolution, python_override=python_override)
+    started = time.monotonic()
+    if dry_run:
+        print(f"[cyan]DRY RUN[/cyan] {' '.join(command)}")
+        return {
+            "project_path": str(plan.project_path),
+            "package_name": plan.package_name,
+            "scenario": scenario_name,
+            "resolution": resolution,
+            "python": python_version,
+            "status": "dry-run",
+            "duration_seconds": 0.0,
+            "payload": None,
+            "error": None,
+        }
+
+    remaining_seconds = deadline - started
+    if remaining_seconds <= 0:
+        return {
+            "project_path": str(plan.project_path),
+            "package_name": plan.package_name,
+            "scenario": scenario_name,
+            "resolution": resolution,
+            "python": python_version,
+            "status": "failed",
+            "duration_seconds": 0.0,
+            "payload": None,
+            "error": "The shared release-validation deadline elapsed before this probe started.",
+        }
+
+    env = dict(os.environ)
+    env.pop("VIRTUAL_ENV", None)
+    try:
+        result = subprocess.run(
+            command,
+            capture_output=True,
+            text=True,
+            timeout=remaining_seconds,
+            check=False,
+            env=env,
+        )
+    except subprocess.TimeoutExpired as exc:
+        stdout = exc.stdout.decode(errors="replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "")
+        stderr = exc.stderr.decode(errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")
+        return {
+            "project_path": str(plan.project_path),
+            "package_name": plan.package_name,
+            "scenario": scenario_name,
+            "resolution": resolution,
+            "python": python_version,
+            "status": "failed",
+            "duration_seconds": round(time.monotonic() - started, 3),
+            "payload": None,
+            "error": f"Release probe exceeded the shared deadline.\n{_truncate_error(stdout, stderr)}",
+        }
+
+    payload = _parse_probe_payload(result.stdout) if result.returncode == 0 else None
+    error = None
+    if result.returncode != 0:
+        error = _truncate_error(result.stdout, result.stderr)
+    elif payload is None:
+        error = "Probe completed without emitting its dependency-version payload."
+
+    return {
+        "project_path": str(plan.project_path),
+        "package_name": plan.package_name,
+        "scenario": scenario_name,
+        "resolution": resolution,
+        "python": python_version,
+        "status": "passed" if error is None else "failed",
+        "duration_seconds": round(time.monotonic() - started, 3),
+        "payload": payload,
+        "error": error,
+    }
+
+
+def _refresh_lockfile(
+    *,
+    workspace_root: Path,
+    deadline: float,
+    dry_run: bool,
+) -> dict[str, Any]:
+    command = ["uv", "lock", "--prerelease", "if-necessary-or-explicit"]
+    if dry_run:
+        print(f"[cyan]DRY RUN[/cyan] {' '.join(command)}")
+        return {"status": "dry-run", "duration_seconds": 0.0, "error": None}
+
+    started = time.monotonic()
+    remaining_seconds = deadline - started
+    if remaining_seconds <= 0:
+        return {
+            "status": "failed",
+            "duration_seconds": 0.0,
+            "error": "The shared release-validation deadline elapsed before uv.lock refresh started.",
+        }
+    try:
+        result = subprocess.run(
+            command,
+            cwd=workspace_root,
+            capture_output=True,
+            text=True,
+            timeout=remaining_seconds,
+            check=False,
+        )
+    except subprocess.TimeoutExpired as exc:
+        stdout = exc.stdout.decode(errors="replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "")
+        stderr = exc.stderr.decode(errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")
+        return {
+            "status": "failed",
+            "duration_seconds": round(time.monotonic() - started, 3),
+            "error": f"uv.lock refresh exceeded the shared deadline.\n{_truncate_error(stdout, stderr)}",
+        }
+
+    error = None if result.returncode == 0 else _truncate_error(result.stdout, result.stderr)
+    return {
+        "status": "passed" if error is None else "failed",
+        "duration_seconds": round(time.monotonic() - started, 3),
+        "error": error,
+    }
+
+
+def run_release_mode(
+    *,
+    workspace_root: Path,
+    base_ref: str,
+    package_filter: str | None,
+    parallelism: int,
+    python_override: str | None,
+    deadline_seconds: int,
+    dry_run: bool,
+    output_json: Path,
+) -> int:
+    """Run fast lower/upper release probes for changed package metadata."""
+    deadline = time.monotonic() + deadline_seconds
+    projects = _build_release_project_map(workspace_root)
+    selected = _selected_release_projects(
+        workspace_root=workspace_root,
+        projects=projects,
+        base_ref=base_ref,
+        package_filter=package_filter,
+    )
+    if not selected:
+        print(f"[red]No changed package pyproject.toml files found relative to {base_ref}.[/red]")
+        return 1
+
+    lock_result = _refresh_lockfile(workspace_root=workspace_root, deadline=deadline, dry_run=dry_run)
+    if lock_result["status"] == "failed":
+        print("[red]uv.lock refresh failed.[/red]")
+        print(f"[red]{lock_result['error']}[/red]")
+        return 1
+
+    plans = [_build_release_probe_plan(workspace_root, project, projects) for project in selected]
+    work_items = [
+        (plan, scenario_name, resolution) for plan in plans for scenario_name, resolution in _RESOLUTION_SCENARIOS
+    ]
+    report: dict[str, Any] = {
+        "started_at": _utc_now(),
+        "mode": "release",
+        "workspace_root": str(workspace_root),
+        "base_ref": base_ref,
+        "python_override": python_override,
+        "deadline_seconds": deadline_seconds,
+        "dry_run": dry_run,
+        "lockfile": lock_result,
+        "packages": [str(plan.project_path) for plan in plans],
+        "probes": [],
+        "summary": {"probes_total": len(work_items), "probes_passed": 0, "probes_failed": 0},
+    }
+    _write_json(output_json, report)
+    print(
+        f"[bold]Running {len(work_items)} lock-independent release probes for {len(plans)} package(s) "
+        f"with a shared {deadline_seconds}s deadline[/bold]"
+    )
+    print(f"[cyan]Writing dependency-bounds release report to {output_json}[/cyan]")
+
+    max_workers = max(1, min(parallelism, len(work_items)))
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+        futures = [
+            executor.submit(
+                _run_release_probe,
+                plan,
+                scenario_name=scenario_name,
+                resolution=resolution,
+                python_override=python_override,
+                deadline=deadline,
+                dry_run=dry_run,
+            )
+            for plan, scenario_name, resolution in work_items
+        ]
+        for future in concurrent.futures.as_completed(futures):
+            result = future.result()
+            report["probes"].append(result)
+            if result["status"] in {"passed", "dry-run"}:
+                report["summary"]["probes_passed"] += 1
+                print(
+                    f"[green]{result['project_path']}: {result['scenario']} passed on Python {result['python']} "
+                    f"({result['duration_seconds']:.1f}s)[/green]"
+                )
+            else:
+                report["summary"]["probes_failed"] += 1
+                print(f"[red]{result['project_path']}: {result['scenario']} failed[/red]")
+                print(f"[red]{result['error']}[/red]")
+            report["updated_at"] = _utc_now()
+            _write_json(output_json, report)
+
+    if report["summary"]["probes_failed"]:
+        print("[bold red]Release dependency-bound validation failed.[/bold red]")
+        return 1
+    print("[bold green]Release dependency-bound validation completed successfully.[/bold green]")
+    return 0
@@ -0,0 +1,217 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import json
+import sys
+from pathlib import Path
+from subprocess import CompletedProcess
+
+import pytest
+
+from scripts.dependencies._dependency_bounds_release_impl import (
+    _PROBE_RESULT_PREFIX,
+    ReleaseProbePlan,
+    _build_release_probe_command,
+    _build_release_probe_plan,
+    _build_release_project_map,
+    _changed_release_project_paths,
+    _parse_probe_payload,
+    run_release_mode,
+)
+from scripts.dependencies.validate_dependency_bounds import main
+
+
+def _write_project(path: Path, content: str) -> None:
+    path.mkdir(parents=True, exist_ok=True)
+    (path / "pyproject.toml").write_text(content)
+
+
+def test_release_probe_uses_only_the_required_internal_dependency_closure(tmp_path: Path) -> None:
+    _write_project(
+        tmp_path,
+        """
+[project]
+name = "agent-framework"
+version = "1.2.0"
+requires-python = ">=3.10"
+dependencies = ["agent-framework-core[all]==1.2.0"]
+
+[tool.uv.workspace]
+members = ["packages/*"]
+
+[tool.flit.module]
+name = "agent_framework_meta"
+""",
+    )
+    _write_project(
+        tmp_path / "packages/core",
+        """
+[project]
+name = "agent-framework-core"
+version = "1.2.0"
+requires-python = ">=3.10"
+dependencies = ["pydantic>=2,<3"]
+
+[project.optional-dependencies]
+all = ["agent-framework-connector>=1,<2"]
+dev = ["pytest>=9"]
+
+[tool.flit.module]
+name = "agent_framework"
+""",
+    )
+    _write_project(
+        tmp_path / "packages/connector",
+        """
+[project]
+name = "agent-framework-connector"
+version = "1.0.0"
+requires-python = ">=3.10"
+dependencies = ["agent-framework-core>=1,<2", "httpx>=0.27,<1"]
+
+[tool.flit.module]
+name = "agent_framework_connector"
+""",
+    )
+    _write_project(
+        tmp_path / "packages/provider",
+        """
+[project]
+name = "agent-framework-provider"
+version = "1.0.0"
+requires-python = ">=3.11"
+dependencies = ["agent-framework-core>=1,<2", "openai>=2,<3"]
+
+[tool.flit.module]
+name = "agent_framework_provider"
+""",
+    )
+
+    projects = _build_release_project_map(tmp_path)
+    provider_plan = _build_release_probe_plan(tmp_path, projects["agent-framework-provider"], projects)
+    provider_editables = "\n".join(provider_plan.editable_specs)
+
+    assert "packages/provider" in provider_editables
+    assert "packages/core" in provider_editables
+    assert "packages/connector" not in provider_editables
+    assert provider_plan.python_version == "3.11"
+
+    root_plan = _build_release_probe_plan(tmp_path, projects["agent-framework"], projects)
+    root_editables = "\n".join(root_plan.editable_specs)
+    assert "packages/core" in root_editables
+    assert "packages/connector" in root_editables
+    assert "pytest" not in root_plan.reported_distributions
+    assert root_plan.python_version == "3.10"
+
+
+def test_release_probe_command_is_lock_independent_and_uses_bound_resolution(tmp_path: Path) -> None:
+    plan = ReleaseProbePlan(
+        project_path=Path("packages/openai"),
+        package_name="agent-framework-openai",
+        editable_specs=(str(tmp_path / "packages/openai"), str(tmp_path / "packages/core")),
+        import_modules=("agent_framework_openai",),
+        reported_distributions=("agent-framework-openai", "openai"),
+        python_version="3.11",
+    )
+
+    command = _build_release_probe_command(plan, resolution="lowest-direct")
+
+    assert "--no-project" in command
+    assert command[command.index("--resolution") + 1] == "lowest-direct"
+    assert command[command.index("--python") + 1] == "3.11"
+    assert command[command.index("--prerelease") + 1] == "if-necessary-or-explicit"
+    assert command.count("--with-editable") == 2
+    assert "pytest" not in command
+    assert "pyright" not in command
+
+    overridden_command = _build_release_probe_command(plan, resolution="highest", python_override="3.12")
+    assert overridden_command[overridden_command.index("--python") + 1] == "3.12"
+
+
+def test_changed_release_projects_are_relative_to_python_workspace(tmp_path: Path, monkeypatch) -> None:
+    def fake_run(*args, **kwargs) -> CompletedProcess[str]:
+        return CompletedProcess(
+            args=args[0],
+            returncode=0,
+            stdout="pyproject.toml\npackages/core/pyproject.toml\nREADME.md\n",
+            stderr="",
+        )
+
+    monkeypatch.setattr("scripts.dependencies._dependency_bounds_release_impl.subprocess.run", fake_run)
+
+    assert _changed_release_project_paths(tmp_path, "upstream/main") == {Path("."), Path("packages/core")}
+
+
+def test_parse_probe_payload_uses_the_last_valid_marker() -> None:
+    first_payload = json.dumps({"versions": {"openai": "2.25.0"}})
+    last_payload = {"imports": ["agent_framework_openai"], "versions": {"openai": "2.47.0"}}
+    stdout = "\n".join((
+        f"{_PROBE_RESULT_PREFIX}{first_payload}",
+        "unrelated subprocess output",
+        f"{_PROBE_RESULT_PREFIX}{json.dumps(last_payload)}",
+    ))
+
+    assert _parse_probe_payload(stdout) == last_payload
+    assert _parse_probe_payload(f"{_PROBE_RESULT_PREFIX}not-json") is None
+    assert _parse_probe_payload(f"{_PROBE_RESULT_PREFIX}[]") is None
+    assert _parse_probe_payload("unrelated subprocess output") is None
+
+
+def test_run_release_mode_dry_run_uses_selected_package_python_floor(tmp_path: Path) -> None:
+    _write_project(
+        tmp_path,
+        """
+[project]
+name = "agent-framework"
+version = "1.2.0"
+requires-python = ">=3.10"
+dependencies = []
+
+[tool.uv.workspace]
+members = ["packages/*"]
+
+[tool.flit.module]
+name = "agent_framework_meta"
+""",
+    )
+    _write_project(
+        tmp_path / "packages/provider",
+        """
+[project]
+name = "agent-framework-provider"
+version = "1.0.0"
+requires-python = ">=3.11"
+dependencies = []
+
+[tool.flit.module]
+name = "agent_framework_provider"
+""",
+    )
+    output_json = tmp_path / "release-results.json"
+
+    exit_code = run_release_mode(
+        workspace_root=tmp_path,
+        base_ref="HEAD",
+        package_filter="provider",
+        parallelism=2,
+        python_override=None,
+        deadline_seconds=300,
+        dry_run=True,
+        output_json=output_json,
+    )
+
+    assert exit_code == 0
+    report = json.loads(output_json.read_text())
+    assert report["python_override"] is None
+    assert report["summary"] == {"probes_total": 2, "probes_passed": 2, "probes_failed": 0}
+    assert {probe["python"] for probe in report["probes"]} == {"3.11"}
+    assert {probe["status"] for probe in report["probes"]} == {"dry-run"}
+
+
+def test_release_mode_rejects_blank_base_ref(monkeypatch, capsys) -> None:
+    monkeypatch.setattr(sys, "argv", ["validate_dependency_bounds", "--mode", "release", "--base-ref", "   "])
+
+    with pytest.raises(SystemExit) as exc_info:
+        main()
+
+    assert exc_info.value.code == 2
+    assert "release mode requires --base-ref" in capsys.readouterr().err
@@ -1,9 +1,10 @@
 # Copyright (c) Microsoft. All rights reserved.
-# ruff: noqa: S404, S603
+# ruff:file-ignore[suspicious-subprocess-import, subprocess-without-shell-equals-true]

 """Unified dependency-bound validation entrypoint.

 Modes:
+- release: run fast lock-independent lower/upper import probes for changed release packages.
 - test: run workspace-wide compatibility gates at lower and upper resolutions.
 - lower: run lower-bound expansion for one package.
 - upper: run upper-bound expansion for one package.
@@ -28,6 +29,7 @@ from pathlib import Path
 import tomli
 from rich import print

+from scripts.dependencies._dependency_bounds_release_impl import run_release_mode
 from scripts.dependencies._dependency_bounds_runtime import (
    extend_command_with_runtime_tools,
    extend_command_with_task,
@@ -363,15 +365,16 @@ def main() -> None:
    """Parse arguments and run the requested dependency-bound mode."""
    parser = argparse.ArgumentParser(
        description=(
-            "Unified dependency-bound workflow. Use mode=test for workspace-wide lower+upper gates, "
+            "Unified dependency-bound workflow. Use mode=release for fast release sanity probes, "
+            "mode=test for the exhaustive workspace lower+upper matrix, "
            "or lower/upper/both for package-scoped or workspace-wide bound expansion."
        )
    )
    parser.add_argument(
        "--mode",
        required=True,
-        choices=("test", "lower", "upper", "both"),
-        help="Execution mode: test (global) or lower/upper/both (package-scoped).",
+        choices=("release", "test", "lower", "upper", "both"),
+        help="Execution mode: release/test gates or lower/upper/both bound expansion.",
    )
    parser.add_argument(
        "--package",
@@ -422,11 +425,49 @@ def main() -> None:
        default="scripts/dependencies/dependency-bounds-test-results.json",
        help="Output report path for test mode.",
    )
+    parser.add_argument(
+        "--base-ref",
+        default=None,
+        help="Git base used to discover changed package metadata in release mode (required unless --package is set).",
+    )
+    parser.add_argument(
+        "--python",
+        default=None,
+        help="Optional Python override for release probes (defaults to each package closure's requires-python floor).",
+    )
+    parser.add_argument(
+        "--release-timeout-seconds",
+        type=int,
+        default=300,
+        help="Shared wall-clock deadline for all release probes.",
+    )
+    parser.add_argument(
+        "--release-output-json",
+        default="scripts/dependencies/dependency-bounds-release-results.json",
+        help="Output report path for release mode.",
+    )
    args = parser.parse_args()

    workspace_root = Path(__file__).resolve().parents[2]
    normalized_package = None if args.package in {None, "", "*"} else args.package

+    if args.mode == "release":
+        base_ref = args.base_ref.strip() if args.base_ref else ""
+        python_override = args.python.strip() if args.python else None
+        if not base_ref and normalized_package is None:
+            parser.error("release mode requires --base-ref unless --package selects one package explicitly")
+        exit_code = run_release_mode(
+            workspace_root=workspace_root,
+            base_ref=base_ref or "HEAD",
+            package_filter=normalized_package,
+            parallelism=args.parallelism,
+            python_override=python_override,
+            deadline_seconds=args.release_timeout_seconds,
+            dry_run=args.dry_run,
+            output_json=(workspace_root / args.release_output_json).resolve(),
+        )
+        raise SystemExit(exit_code)
+
    if args.mode == "test":
        exit_code = _run_test_mode(
            workspace_root=workspace_root,
Author	SHA1	Message	Date
Evan Mattson	cd5d282827	Merge branch 'main' into feature/python-subworkflow-checkpoint	2026-07-24 12:25:13 +09:00
Tao Chen	535690cd1d	Add warning	2026-07-23 15:25:58 -07:00
Evan Mattson	e90b6de5a7	Python: Improve python package management operations (#7274 ) * improve package mgmt timings * Address Python release validation review feedback	2026-07-23 22:14:10 +00:00
Atharva Vichare	d98ac29115	Python: Fix duplicate function call on approval round-trip (#7267 ) (#7271 ) * Python: Fix duplicate function call on approval round-trip (#7267) `_replace_approval_contents_with_results` deduped restored function calls against only the message currently being scanned. On an approval round-trip the hosting layer replays the stored `function_call` item and its `mcp_approval_request` item as two separate assistant messages, so the per-message check never fired and the approval request restored a second copy of the call. Only one copy received the function result; the orphaned copy was left unanswered, which the Responses API rejects with "No tool output found for function call call_<id>". Collect existing call ids across all messages instead, and add a restored call to that set so two approval requests for the same call cannot both expand. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * Refactor approval placeholder result handling Refactor approval handling logic to improve clarity and maintainability. * Refactor test to support reused call IDs after completion Updated the test to allow reused call IDs after completion, ensuring that a completed call does not suppress later approval requests with the same ID. Adjusted assertions to reflect the new behavior. --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>	2026-07-23 21:27:26 +00:00
Tao Chen	85fde62a76	Fix syntax check	2026-07-20 14:14:24 -07:00
Tao Chen	85eb53d412	Address comments	2026-07-20 13:27:35 -07:00
Tao Chen	2d7c8da6b0	Merge branch 'main' into feature/python-subworkflow-checkpoint	2026-07-20 11:16:36 -07:00
Tao Chen	e0b0b79d9e	Do not allow checkpoint storage in sub workflow	2026-07-14 14:30:46 -07:00
Tao Chen	a4f6c26990	Clean up	2026-07-14 11:28:48 -07:00
Tao Chen	1389f304f2	Drop redundant decode in WorkflowExecutor.on_checkpoint_restore The storage backend already materializes the full checkpoint on load (FileCheckpointStorage decodes recursively; InMemoryCheckpointStorage deep-copies), so the embedded sub_workflow_checkpoint (and legacy execution_contexts) arrive already decoded - like every other executor's on_checkpoint_restore state. Remove the no-op decode_checkpoint_value calls and the now-unused import.	2026-07-14 11:11:49 -07:00
Tao Chen	93719f4a34	Remove per-execution bookkeeping from WorkflowExecutor The sub-workflow is a single shared instance, so per-execution ExecutionContext/request routing never provided real isolation. Delegate request/response tracking to the sub-workflow itself: can_handle accepts targeted propagated responses, _handle_response validates against the sub-workflow's pending requests and forwards responses immediately, and on_checkpoint_save embeds only the sub-workflow checkpoint (on_checkpoint_restore keeps a legacy reader for older checkpoints). Also emit the fresh-message/checkpoint-while-pending warning from FunctionalWorkflow.run to match Workflow.run.	2026-07-14 10:46:14 -07:00
Tao Chen	a486374fd8	Move checkpoint-object construction into the runner context Add RunnerContext.create_checkpoint_object alongside create_checkpoint (create_checkpoint now delegates to it and persists), so Runner.capture_checkpoint_object builds the snapshot via the context instead of a one-off get_messages peek primitive. In-flight messages are captured non-destructively (per-source lists copied). The checkpoint-less capturing contexts (azurefunctions, durabletask) raise NotImplementedError to match create_checkpoint.	2026-07-13 15:57:37 -07:00
Tao Chen	e78604103d	Fix sub-workflow checkpoint restore to preserve sub-workflow state Add Runner.capture_checkpoint_object/restore_from_checkpoint_object (quiescent-only nested checkpoint) and embed a sub_workflow_checkpoint in WorkflowExecutor.on_checkpoint_save/on_checkpoint_restore so a resumed parent restores each sub-workflow's mid-progress state instead of only replaying pending request-info events. Keeps a backward-compat fallback when sub_workflow_checkpoint is absent.	2026-07-13 15:18:53 -07:00