From d2a9ec85fe655641eac3be2d2b387b97cd9c9835 Mon Sep 17 00:00:00 2001 From: Viren Baraiya Date: Sun, 9 Aug 2026 13:31:12 -0700 Subject: [PATCH 1/9] conductor workspace baseline [conductor-workspace:e6daf42f-b15a-4630-a777-20333422f915:baseline] Conductor-Original-Branch: main Conductor-Original-Head: 61dc1795719c58f9ad7f722cc3d30692e9a89478 From 9b3a2a8ff138f0ae2eb1604422a786d2e5cfb4e7 Mon Sep 17 00:00:00 2001 From: Viren Baraiya Date: Sun, 9 Aug 2026 13:40:03 -0700 Subject: [PATCH 2/9] approved design docs (docs/design) --- docs/design/AGENT_SDK_PORTING_SPEC.md | 46 +- .../design/WORKER_SDK_IMPLEMENTATION_GUIDE.md | 27 +- docs/design/architecture.md | 418 ++++++++++++++++++ 3 files changed, 476 insertions(+), 15 deletions(-) create mode 100644 docs/design/architecture.md diff --git a/docs/design/AGENT_SDK_PORTING_SPEC.md b/docs/design/AGENT_SDK_PORTING_SPEC.md index 51200e3b..987e6261 100644 --- a/docs/design/AGENT_SDK_PORTING_SPEC.md +++ b/docs/design/AGENT_SDK_PORTING_SPEC.md @@ -437,13 +437,24 @@ The runtime verbs MUST behave exactly as follows: be **reconstructible in the executor context** that runs it. - If your worker executors are **spawned processes** (Python-style): callables - must be importable by qualified name or serializable by value — module-level - functions, or module-level classes instantiated with plain-data fields. Local - closures, lambdas capturing live objects, and functions defined inside other - functions MUST be rejected at registration time with an actionable error - ("define the callable at module level"). Entry-point scripts must guard - top-level orchestration with the language's main-module idiom so a re-import in - the child does not re-run it. + must be importable by qualified name, deterministically reconstructible from + an importable decorator container, or serializable by value. Module-level + decorator functions remain valid when the decorator rebinds the public name: + support `__wrapped__`, stable container attributes such as `func` and + `coroutine`, and deterministic bounded nested/closure traversal for containers + such as OpenAI Agents SDK `FunctionTool`. Parent-side discovery and child-side + reconstruction MUST share the traversal implementation. Discovery retains all + importable plain-function candidates and selects only by the exact tool-name / + sole-candidate contract; it MUST NOT inspect signatures or exclude valid + `ctx`/`context` first parameters. Stable `FunctionKey` code identity is used + only to prove reconstruction of that already-selected function in the child. + Plain functions never silently fall back + to direct-object pickling; local functions, lambdas, and bound methods fail at + registration. Direct callable objects require a standard-pickle round trip. + Entry-point scripts must guard top-level orchestration with the language's + main-module idiom so a re-import in the child does not re-run it. The exact + Python types, traversal order, failure semantics, and version-pinned tests are + authoritative in [`architecture.md`](architecture.md). - If your executors are **threads in-process** (Java/Go/C#/TS typical): the invariant reduces to "no capture of per-run mutable runtime state"; document it, and keep factory boundaries clean regardless (next point). @@ -458,8 +469,25 @@ be **reconstructible in the executor context** that runs it. **Acceptance criteria** - [ ] Registering a closure/lambda as a tool fails fast with an actionable error (process-spawn runtimes) or is impossible by API design (typed runtimes). -- [ ] A registration-time round-trip test: serialize/reconstruct each registered - callable the way the executor would, and invoke it. +- [ ] A module-level OpenAI Agents SDK `@function_tool` registers under `spawn` + even though the module global resolves to a `FunctionTool` container; + Python regression coverage uses the repository-local + `tests/requirements/openai_agents_0_18_2.txt` constraint, verifies the + installed version, and accepts either the deterministic `unwrap_depth` or + deep-extraction strategy. +- [ ] Valid decorated tools whose first parameter is named `context` and `ctx` + each traverse the production registration path and execute the intended + named callable in a real spawned child; parameter names never filter or + rank embedded-function candidates. +- [ ] A rebound global with no supported reconstruction path reports the rebound + container mismatch instead of incorrectly telling the user to move an + already module-level function to module scope. +- [ ] A registration-time round-trip test serializes through the framework + serializer into `WorkerInfo` records, passes those records to + `_register_framework_workers`, captures the registered worker through + `get_registered_workers()`, then reconstructs and invokes it in a real + spawned child. Exact fixtures and commands are defined in + [`architecture.md`](architecture.md). --- diff --git a/docs/design/WORKER_SDK_IMPLEMENTATION_GUIDE.md b/docs/design/WORKER_SDK_IMPLEMENTATION_GUIDE.md index 73eba35c..3e841261 100644 --- a/docs/design/WORKER_SDK_IMPLEMENTATION_GUIDE.md +++ b/docs/design/WORKER_SDK_IMPLEMENTATION_GUIDE.md @@ -2540,12 +2540,27 @@ wrapped as a **passthrough worker**; its tools run as Conductor workers; lifecyc tool-use events are pushed to the server for observability. **Spawn-safety (critical).** Workers execute in spawned processes, so every worker/tool -callable must be **importable by qualified name or picklable by value** — a module-level -function, or a module-level callable *class instance* holding only plain data. **Never** -register a `` closure or a lambda. Entry scripts must guard top-level execution -with the language's "main module" guard (e.g. `if __name__ == "__main__":`) so a -re-imported spawn child does not re-run the orchestration. Framework worker factories -receive plain strings (server URL + credentials), never live client objects. +callable must be **importable by qualified name, deterministically reconstructible from +an importable decorator container, or picklable by value**. Decorators that rebind a +module global do not make the original module-level function unsafe: reconstruction +must support `__wrapped__`, stable `func`/`coroutine` container attributes, and a +deterministic bounded nested/closure traversal for containers such as OpenAI Agents +SDK `FunctionTool`. Discovery and reconstruction must share one implementation and +retain every importable plain-function candidate. Parent-side selection uses only the +exact tool-name / sole-candidate contract; it must not inspect signatures or exclude +valid first parameters named `ctx` or `context`. Stable `FunctionKey` code identity is +used only to prove reconstruction of the already-selected function in the child. +Plain functions must never fall +back to direct-object pickling; direct callable objects require a standard-pickle +round trip. **Never** register a true `` closure, lambda, or bound method. +Entry scripts must guard top-level execution with the language's "main module" guard (e.g. +`if __name__ == "__main__":`) so a re-imported spawn child does not re-run the +orchestration. Framework worker factories receive plain strings (server URL + +credentials), never live client objects. The exact Python types, resolution order, +traversal algorithm, failure semantics, repository-local +`openai-agents==0.18.2` constraint, and production-path `WorkerInfo` registration test +including the required `context` and `ctx` spawn regressions are defined in +[`architecture.md`](architecture.md). ### 25.8 Agent Credentials (runtimeMetadata contract) diff --git a/docs/design/architecture.md b/docs/design/architecture.md new file mode 100644 index 00000000..da6c184b --- /dev/null +++ b/docs/design/architecture.md @@ -0,0 +1,418 @@ +# Agent Worker Callable Transport Architecture + +**Status:** Authoritative + +**Last updated:** 2026-08-09 + +**Scope:** Python agent-framework tool registration across multiprocessing +`spawn` and `forkserver` executors. + +This document is the source of truth for the complete file layout, shared +contracts, exact types, data model, and naming conventions used to transport +agent tool callables into worker processes. It resolves GitHub issue #448 +without changing the SDK multiprocessing start method or weakening +registration-time validation. + +## Problem Statement + +Python workers use `spawn` on every supported operating system. A worker +callable must cross a standard-`pickle` boundary and be reconstructible after +the child imports its defining module. + +Some decorators replace the module global that originally named the function: + +- LangChain `@tool` binds the global to a `StructuredTool` and exposes the + original through `func` or `coroutine`. +- OpenAI Agents SDK `@function_tool` binds the global to a `FunctionTool`. In + the issue's reproduced `openai-agents==0.18.2` shape, the original function + is reachable through nested invocation state and a closure rather than a + supported public attribute. + +The original function is module-level, but normal function pickling fails +because importing `module.get_inventory` returns the decorator container, not +the same function object. This differs from issue #443: a true local contains +`` in its qualified name and remains unsupported by this focused fix. + +## Compatibility Boundary + +The regression contract is pinned to `openai-agents==0.18.2`, the version from +issue #448. The pin is test-fixture-only: + +```text +tests/requirements/openai_agents_0_18_2.txt +``` + +contains exactly: + +```text +openai-agents==0.18.2 +``` + +The targeted regression environment MUST be installed with: + +```bash +python -m pip install -e ".[openai-agents]" \ + -c tests/requirements/openai_agents_0_18_2.txt +``` + +The regression test MUST also assert +`importlib.metadata.version("openai-agents") == "0.18.2"` before constructing +the fixture. This makes an incorrectly prepared environment fail explicitly +rather than silently testing another framework shape. + +This fixture does not change `pyproject.toml`, `poetry.lock`, the existing +`.[agents]` install path, or any GitHub Actions workflow. The package's public +optional dependency remains `openai-agents>=0.12.2`; the existing lockfile may +continue selecting its current version. Newer OpenAI Agents SDK versions MAY +resolve through a simpler `__wrapped__` path, but they must not replace the +`0.18.2` regression fixture. + +This design intentionally keeps standard `pickle`. It does not add +`cloudpickle` transport for local functions, closures, or lambdas. + +## Authoritative File Layout + +```text +src/conductor/ai/agents/ +├── frameworks/ +│ └── serializer.py +│ ├── discovers framework tools +│ └── emits WorkerInfo.func as the original plain function +└── runtime/ + ├── _worker_entries.py + │ ├── SpawnSafetyError + │ ├── FunctionKey + │ ├── FunctionRef + │ ├── ToolWorkerEntry + │ ├── _iter_embedded_functions + │ ├── wrap_callable / unwrap_callable + │ └── probe_spawn_safety + └── runtime.py + └── _register_framework_workers + +src/conductor/client/automator/ +└── task_handler.py + └── establishes the multiprocessing start method used by workers + +tests/unit/ +├── ai/ +│ ├── test_worker_entries.py +│ └── test_openai_agents_spawn_registration.py +└── resources/ + └── openai_agents_entry_helpers.py + +tests/requirements/ +└── openai_agents_0_18_2.txt +``` + +`frameworks/serializer.py` and `runtime/_worker_entries.py` MUST call the same +`_iter_embedded_functions` implementation. Parent-side extraction and +child-side reconstruction MUST NOT have parallel traversal implementations. + +## Data Model + +### `FunctionKey` + +`FunctionKey` identifies a plain function across parent and spawned-child +imports without depending on object identity or a framework-private attribute +path. + +```python +@dataclass(frozen=True) +class FunctionKey: + module: str + qualname: str + code_sha256: str +``` + +`code_sha256` is the lowercase SHA-256 hex digest of +`marshal.dumps(fn.__code__)`. `FunctionKey.of(fn)` accepts only +`inspect.isfunction(fn)` values. The code digest distinguishes an original +function from wrappers that copied its `__module__` and `__qualname__` with +`functools.wraps`. + +### `FunctionRef` + +`FunctionRef` is an immutable, pickle-safe recipe for recovering one plain +function after importing its module. + +```python +@dataclass(frozen=True) +class FunctionRef: + module: str + qualname: str + unwrap_depth: int = 0 + attr_hop: str = "" + deep_extract: bool = False + expected_key: FunctionKey | None = None +``` + +| Field | Contract | +|---|---| +| `module` | Importable module containing the public global. | +| `qualname` | Attribute path from the module to the public global. It MUST NOT contain `` or ``. | +| `unwrap_depth` | Number of `__wrapped__` hops after resolving the public global or `attr_hop`. Range: `0..32`. | +| `attr_hop` | One stable container attribute. Supported values are `""`, `"func"`, and `"coroutine"`. | +| `deep_extract` | Whether reconstruction uses bounded embedded-function traversal. | +| `expected_key` | Required only for deep extraction; identifies the exact candidate selected in the parent. | + +Invariants: + +- `attr_hop` and `deep_extract` MUST NOT both be active. +- `deep_extract=True` requires `unwrap_depth=0` and a non-null `expected_key`. +- `deep_extract=False` requires `expected_key=None`. + +### `ToolWorkerEntry` + +`ToolWorkerEntry` transports a tool target plus plain worker metadata: + +```python +ToolWorkerEntry( + tool_name: str, + fn_ref: FunctionRef | None = None, + fn_direct: Callable | None = None, + guardrails: list | None = None, + credential_names: list[str] | None = None, + framework_callable: bool = False, +) +``` + +Exactly one of `fn_ref` or `fn_direct` MUST be set. + +- `fn_ref` is mandatory for every plain Python function, including a function + recovered from a decorator container. +- `fn_direct` is reserved for non-function callable objects that pass the + direct-object round-trip contract below. Putting a plain function in + `fn_direct` is forbidden because standard `pickle` still serializes it by + module and qualified name. + +## Deterministic Embedded-Function Traversal + +`_iter_embedded_functions(root, max_depth=2)` yields candidate plain functions +in a deterministic breadth-first traversal. It is framework-agnostic and MUST +NOT hardcode OpenAI private attribute names or closure-cell positions. + +The algorithm is exact: + +1. Maintain a FIFO queue of `(object, attribute_depth)`, initialized with + `(root, 0)`, plus `visited_object_ids` and `yielded_function_ids` sets. +2. Pop FIFO. If its identity was visited, skip it; otherwise mark it visited. +3. If the object is a plain function: + 1. Read `__closure__` without evaluating descriptors. + 2. Visit closure cells in numeric index order. Ignore empty cells + (`ValueError`). For each cell containing a plain function, emit it first + if it is an eligible candidate and has not already been emitted. + 3. Emit the function itself if eligible and not already emitted. + 4. Do not enqueue arbitrary function attributes. +4. If the object is not a function and `attribute_depth < max_depth`, obtain + `vars(object)`. If `vars` raises `Exception` (but not `BaseException`), treat + the object as a leaf. Iterate the resulting mapping's keys in + lexicographic order; read values directly from the mapping, never with + `getattr`, so properties and descriptors cannot execute. Enqueue non-scalar, + non-module, non-type values at `attribute_depth + 1`. +5. Scalars (`None`, booleans, numbers, strings, bytes), modules, and classes are + leaves. Cycles are harmless because identities are visited once. + +A function is an eligible candidate when all of these hold: + +- `inspect.isfunction(candidate)` is true; +- `candidate.__module__` is a non-empty string; +- `candidate.__qualname__` is a non-empty string containing neither `` + nor ``. + +Traversal MUST NOT inspect a candidate's signature or exclude it based on +parameter names. In particular, `ctx` and `context` are valid first-parameter +names for context-aware tools and MUST remain in the candidate set. + +Closure inspection does not consume attribute depth. Therefore the reproduced +`0.18.2` shape may use two object-attribute edges and then inspect the reached +wrapper function's closure. + +### Candidate selection and proof + +Traversal never means "take the first plausible function." + +- In the parent, framework serializer extraction retains every eligible + candidate and deduplicates repeated references to the same object. It selects + by the existing tool-name/uniqueness contract only: first filter by exact + `candidate.__name__ == tool.name`; exactly one match selects that candidate. + If there are no name matches, the sole eligible candidate may be selected. + More than one name match, or more than one eligible candidate when there is + no name match, is ambiguous and raises `SpawnSafetyError` listing candidate + keys in traversal order. Signatures, annotations, parameter names, and + `FunctionKey` values MUST NOT rank, filter, or select parent-side candidates. +- `FunctionRef.of(target)` uses object identity to confirm that deep traversal + of the rebound public global contains `target`, then stores + `expected_key=FunctionKey.of(target)`. +- In the child, `FunctionRef.resolve()` repeats the same traversal and selects + candidates whose `FunctionKey` equals `expected_key`. Exactly one distinct + candidate object must match. Zero matches means the imported definition + changed; multiple matches mean reconstruction is ambiguous. Both cases raise + `SpawnSafetyError` before invocation. + +`FunctionKey` is used only after parent-side selection, to prove that child +reconstruction recovered the same source callable rather than a neighboring +closure or framework wrapper. It is not a discovery or selection heuristic. + +## `FunctionRef` Resolution Contract + +`FunctionRef.of(fn)` accepts only a plain function and selects the first +successful deterministic strategy in this order: + +1. **Direct identity:** resolving `module + qualname` returns `fn`. +2. **Wrapped function:** following at most 32 `__wrapped__` hops from the public + global reaches `fn`; record `unwrap_depth`. +3. **Stable container attribute:** `func` then `coroutine`, in that order, + reaches `fn`, optionally followed by at most 32 `__wrapped__` hops; record + `attr_hop` and `unwrap_depth`. +4. **Deep extraction:** deterministic traversal of the public global contains + `fn`; set `deep_extract=True` and record `expected_key`. +5. **Failure:** raise `SpawnSafetyError` saying the public name resolves to a + rebound container but no supported path reconstructs the requested function. + +The OpenAI `FunctionTool` acceptance condition is strategy-neutral. Depending +on the exact framework object shape, a module-level `@function_tool` MAY encode +as a positive `unwrap_depth` or as `deep_extract=True`; tests MUST assert +successful deterministic reconstruction, not require one contradictory flag. + +`FunctionRef.resolve()` imports `module`, walks `qualname`, performs only the +encoded strategy, validates the final object is a plain function, and caches it +per process. Attribute and unwrap failures are converted to `SpawnSafetyError` +that names the module, qualified name, and failed strategy. + +## `ToolWorkerEntry.for_callable` Contract + +`ToolWorkerEntry.for_callable(fn, tool_name, ...)` follows these rules exactly: + +1. If `inspect.isfunction(fn)`, call `FunctionRef.of(fn)`. On failure, propagate + its `SpawnSafetyError`. A plain function MUST NOT fall back to `fn_direct`. + This covers module-level rebound functions, true locals, and lambdas. +2. If `inspect.ismethod(fn)`, reject it with `SpawnSafetyError`: bound methods + are unsupported because their instance transport and rebinding semantics are + not the callable-object contract. The error instructs the user to expose a + module-level function or a module-level callable class instance. +3. If `fn` is any other callable object, validate direct transport by running + `payload = pickle.dumps(fn)` and `clone = pickle.loads(payload)` with standard + `pickle`. The clone MUST be callable, and `pickle.dumps(clone)` MUST also + succeed. Only then construct `fn_direct=fn`. +4. Non-callable values are rejected with `SpawnSafetyError`. + +Direct-object validation catches and wraps the original exception. Its message +MUST name `tool_name`, the callable type, and state that standard-pickle +round-trip validation failed. It MUST NOT recommend module scope when the +actual failure is unpickleable object state. + +Required error behavior: + +| Input | Result | +|---|---| +| Module-level plain function, directly importable | `fn_ref` | +| Module-level decorator-rebound function with a supported path | `fn_ref` | +| Local function | `SpawnSafetyError` naming ``; no direct fallback | +| Lambda | `SpawnSafetyError` naming ``; no direct fallback | +| Bound method | `SpawnSafetyError` naming bound methods as unsupported | +| Module-level callable instance with pickle-safe state | `fn_direct` after two serialization passes and one load | +| Callable instance with unpickleable state | `SpawnSafetyError` with the round-trip cause | +| Rebound decorator container that cannot reconstruct the requested function | `SpawnSafetyError` explaining the container mismatch, not "move it to module level" | + +Framework serialization MUST emit the extracted original plain function in +`WorkerInfo.func`; it MUST NOT pass the `FunctionTool`/`StructuredTool` +container itself to `for_callable` as a way to bypass reconstruction. + +## Registration Contract + +`AgentRuntime._register_framework_workers` remains the production path: + +1. `frameworks.serializer.serialize_agent()` emits `WorkerInfo` records. +2. `make_tool_worker()` creates a `ToolWorkerEntry` through `for_callable`. +3. `probe_spawn_safety(wrapper, worker_info.name, group="tools")` constructs a + throwaway Conductor `Worker` around that exact `ToolWorkerEntry` and + standard-pickles the complete `Worker` whenever the active start method is + `spawn` or `forkserver`. +4. Only after the probe succeeds does `worker_task(...)(wrapper)` call + `register_decorated_fn`, which stores the exact entry at + `_decorated_functions[(worker_info.name, None)]["func"]`. +5. `get_registered_workers()` materializes the production Conductor `Worker` + from that registry record. Tests use this accessor instead of reconstructing + a lookalike `Worker` themselves. + +This applies on Linux and macOS. The fix MUST NOT force `fork`, add platform +checks, or restore a multiprocessing start-method override. + +Errors for an unreconstructible rebound global MUST state that the already +module-level public name resolves to a decorator container without a supported +`__wrapped__`, `func`/`coroutine`, or deterministic embedded-function match. + +## Verification Contract + +The regression suite MUST use real `openai-agents==0.18.2` objects defined in +the importable `tests/unit/resources/openai_agents_entry_helpers.py` module: + +- one synchronous module-level `@function_tool` matching the issue's + `get_inventory(sku: str) -> str` reproduction; +- one valid module-level decorated context-aware tool whose first parameter is + named exactly `context`; +- one valid module-level decorated context-aware tool whose first parameter is + named exactly `ctx`; +- the decorated global left bound to the real `FunctionTool` container; +- a module-level spawn-child target that unpickles a complete Conductor + `Worker`, creates a real `Task` with the fixture's input, calls + `Worker.execute(task)`, and returns status plus output through a queue. + +`tests/unit/ai/test_openai_agents_spawn_registration.py` MUST exercise the +production path rather than isolated `FunctionRef` construction: + +1. Assert the installed distribution version is exactly `0.18.2`. +2. Put the real tool on a real OpenAI Agents SDK `Agent` and call + `frameworks.serializer.serialize_agent()`, producing + `(raw_config, workers: list[WorkerInfo])`. +3. Assert `len(workers) == 1`, `workers[0].name == "get_inventory"`, and + `inspect.isfunction(workers[0].func)`. The value passed to registration is + this serialized `WorkerInfo` list, never the `Agent` object. +4. Create a minimal `AgentRuntime` receiver with + `AgentConfig(auto_start_workers=False)`. This disables worker-manager polling + only; it does not replace registration. Remove any stale + `_decorated_functions[("get_inventory", None)]` record before the call. +5. Call `runtime._register_framework_workers(workers)`. Do not patch + `worker_task`, `register_decorated_fn`, `make_tool_worker`, or + `probe_spawn_safety`. Optional spies may wrap the latter two with + `unittest.mock.patch(..., wraps=real_function)` solely to assert each real + implementation ran once. +6. Read `_decorated_functions[("get_inventory", None)]["func"]` and assert it + is the actual `ToolWorkerEntry` registered by `worker_task`, with `fn_ref` + set and `fn_direct is None`. Then call `get_registered_workers()` and select + the `Worker` whose `task_definition_name == "get_inventory"`; assert its + `execute_function` is that same entry. This is the precise capture point for + the production Conductor worker. +7. Standard-pickle round-trip that complete `Worker`, then pass its bytes to a + real `multiprocessing.get_context("spawn")` child using the helper target. + Assert exit code zero, `COMPLETED`, and + `output_data == {"result": "SKU ABC-123: 42 units"}`. +8. Delete `_decorated_functions[("get_inventory", None)]` in `finally` so the + process-wide registry cannot leak into other unit tests. + +Repeat the same serializer-to-registration-to-real-spawn-child path for the +decorated `context` and `ctx` fixtures. Each fixture MUST return a distinct +sentinel result, and the assertion MUST prove that the intended named callable +executed. These tests are regression guards against signature-name filtering +and against selecting a neighboring closure candidate. They MUST fail if +either first-parameter name is excluded, ignored in favor of another embedded +function, or reconstructed as a different callable. + +The executable targeted verification command is: + +```bash +python -m pytest tests/unit/ai/test_openai_agents_spawn_registration.py -q +``` + +It is run after the constrained install above. Existing broad unit-test +commands remain unchanged and are not responsible for selecting `0.18.2`. + +The suite MUST retain negative coverage for locals, lambdas, bound methods, +callable instances with unpickleable state, ambiguous deep traversal, changed +child definitions (zero key matches), direct references, `__wrapped__`, and +LangChain `func`/`coroutine` reconstruction. + +The Conductor UI OpenAI quickstart lives in the separate +`conductor-oss/conductor` repository. Adding a tool to that sample is a +cross-repository follow-up and is outside this repository's change scope. From 8eab46ab16426de5acd57420aaff6337999c7ab5 Mon Sep 17 00:00:00 2001 From: Viren Baraiya Date: Sun, 9 Aug 2026 13:42:01 -0700 Subject: [PATCH 3/9] code_subtask share-framework-function-discovery --- src/conductor/ai/agents/frameworks/serializer.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/conductor/ai/agents/frameworks/serializer.py b/src/conductor/ai/agents/frameworks/serializer.py index f455ce96..be698908 100644 --- a/src/conductor/ai/agents/frameworks/serializer.py +++ b/src/conductor/ai/agents/frameworks/serializer.py @@ -326,9 +326,8 @@ def _try_extract_tool_object(obj: Any) -> Optional[WorkerInfo]: description = getattr(obj, "description", None) or "" - # Find the original callable by searching the object's attribute tree - # (up to 2 levels deep) for a plain function - original_func = _find_embedded_function(obj, max_depth=2) + # Use the same traversal and depth policy as spawned-child reconstruction. + original_func = _find_embedded_function(obj) if original_func is None: # No callable found — still emit as a tool but without a local worker logger.debug("Tool-like object '%s' has no extractable callable", name) From f8a75e6f8908d427ec56c95a4010833c5274b75d Mon Sep 17 00:00:00 2001 From: Viren Baraiya Date: Sun, 9 Aug 2026 13:42:06 -0700 Subject: [PATCH 4/9] code_subtask cover-openai-function-tool-spawn --- tests/unit/ai/test_worker_entries.py | 69 ++++++++++++------- .../resources/openai_agents_entry_helpers.py | 21 ++++++ 2 files changed, 67 insertions(+), 23 deletions(-) diff --git a/tests/unit/ai/test_worker_entries.py b/tests/unit/ai/test_worker_entries.py index 45124043..f21a9927 100644 --- a/tests/unit/ai/test_worker_entries.py +++ b/tests/unit/ai/test_worker_entries.py @@ -167,59 +167,82 @@ def test_cross_process_spawn_roundtrip_with_hop(self): class TestFunctionRefDeepExtract: """OpenAI Agents SDK 0.19+ exposes FunctionTool functions via wrappers.""" - def test_sync_function_tool_deep_extract(self): + @staticmethod + def _openai_tools(): pytest.importorskip("agents") from tests.unit.resources import openai_agents_entry_helpers as oa - raw = _find_embedded_function(oa.oa_get_weather) + return oa, _find_embedded_function(oa.oa_get_weather), _find_embedded_function( + oa.oa_get_weather_async + ) + + def test_sync_function_tool_deep_extract(self): + oa, raw, _ = self._openai_tools() assert raw is not None ref = FunctionRef.of(raw) assert ref == FunctionRef(oa.__name__, "oa_get_weather", unwrap_depth=1) assert ref.resolve() is raw def test_async_function_tool_deep_extract(self): - pytest.importorskip("agents") - from tests.unit.resources import openai_agents_entry_helpers as oa - - raw = _find_embedded_function(oa.oa_get_weather_async) + _, _, raw = self._openai_tools() assert raw is not None ref = FunctionRef.of(raw) assert ref.unwrap_depth == 1 assert ref.deep_extract is False assert ref.resolve() is raw - def test_ref_pickles(self): - pytest.importorskip("agents") - from tests.unit.resources import openai_agents_entry_helpers as oa - - raw = _find_embedded_function(oa.oa_get_weather) + @pytest.mark.parametrize("is_async", [False, True]) + def test_function_ref_pickles(self, is_async): + _, sync_raw, async_raw = self._openai_tools() + raw = async_raw if is_async else sync_raw + assert raw is not None ref = pickle.loads(pickle.dumps(FunctionRef.of(raw))) assert ref.resolve() is raw - def test_entry_transports_function_tool_fn_by_ref(self): + @pytest.mark.parametrize("is_async", [False, True]) + def test_entry_transports_function_tool_fn_by_ref(self, is_async): # Pre-fix this fell to fn_direct, whose reference pickling then found # the FunctionTool at the global name: "it's not the same object as …". - pytest.importorskip("agents") - from tests.unit.resources import openai_agents_entry_helpers as oa - - raw = _find_embedded_function(oa.oa_get_weather) - entry = ToolWorkerEntry.for_callable(raw, "oa_get_weather") + _, sync_raw, async_raw = self._openai_tools() + raw = async_raw if is_async else sync_raw + assert raw is not None + entry = ToolWorkerEntry.for_callable( + raw, "oa_get_weather_async" if is_async else "oa_get_weather" + ) assert entry.fn_ref is not None clone = pickle.loads(pickle.dumps(entry)) assert clone._target() is raw - def test_cross_process_spawn_roundtrip(self): - pytest.importorskip("agents") - from tests.unit.resources import openai_agents_entry_helpers as oa + def test_sync_function_tool_entry_executes_in_spawn_child(self): + _, raw, _ = self._openai_tools() + assert raw is not None + entry_bytes = pickle.dumps(ToolWorkerEntry.for_callable(raw, "oa_get_weather")) + + ctx = multiprocessing.get_context("spawn") + q = ctx.Queue() + p = ctx.Process( + target=oa.run_weather_entry_child, + args=(entry_bytes, "Boston", q), + ) + p.start() + try: + status, output = q.get(timeout=30) + finally: + p.join(timeout=30) + assert p.exitcode == 0 + assert "COMPLETED" in status + assert output == {"result": "sunny in Boston"} - raw = _find_embedded_function(oa.oa_get_weather) + def test_async_function_tool_ref_executes_in_spawn_child(self): + _, _, raw = self._openai_tools() + assert raw is not None ctx = multiprocessing.get_context("spawn") q = ctx.Queue() ref_bytes = pickle.dumps(FunctionRef.of(raw)) - p = ctx.Process(target=helpers.resolve_and_call_child, args=(ref_bytes, "Boston", q)) + p = ctx.Process(target=oa.resolve_and_await_child, args=(ref_bytes, "Boston", q)) p.start() try: - assert q.get(timeout=30) == "sunny in Boston" + assert q.get(timeout=30) == "async sunny in Boston" finally: p.join(timeout=30) assert p.exitcode == 0 diff --git a/tests/unit/resources/openai_agents_entry_helpers.py b/tests/unit/resources/openai_agents_entry_helpers.py index 18524a89..2070d92a 100644 --- a/tests/unit/resources/openai_agents_entry_helpers.py +++ b/tests/unit/resources/openai_agents_entry_helpers.py @@ -6,6 +6,9 @@ openai-agents extra can still import that module; tests importing THIS module must be gated with ``pytest.importorskip("agents")``. """ +import asyncio +import pickle + from agents import function_tool @@ -19,3 +22,21 @@ def oa_get_weather(city: str) -> str: async def oa_get_weather_async(city: str) -> str: """Return a canned weather string for a city, asynchronously.""" return f"async sunny in {city}" + + +def resolve_and_await_child(ref_bytes: bytes, city: str, q) -> None: + """Spawn-child target: unpickle a FunctionRef and await its function.""" + ref = pickle.loads(ref_bytes) + q.put(asyncio.run(ref.resolve()(city))) + + +def run_weather_entry_child(entry_bytes: bytes, city: str, q) -> None: + """Spawn-child target: run the sync weather tool as a real task.""" + from conductor.client.http.models import Task + + entry = pickle.loads(entry_bytes) + task = Task(task_id="t-openai-spawn-1", workflow_instance_id="wf-openai-spawn-1") + task.input_data = {"city": city} + task.task_def_name = entry.tool_name + result = entry(task) + q.put((str(result.status), dict(result.output_data or {}))) From a63d9ef9649da2d6ab9afa2441b3f0f69c8773c4 Mon Sep 17 00:00:00 2001 From: Viren Baraiya Date: Sun, 9 Aug 2026 13:42:20 -0700 Subject: [PATCH 5/9] code_subtask support-deep-function-references --- .../ai/agents/runtime/_worker_entries.py | 248 ++++++++++++------ 1 file changed, 166 insertions(+), 82 deletions(-) diff --git a/src/conductor/ai/agents/runtime/_worker_entries.py b/src/conductor/ai/agents/runtime/_worker_entries.py index 044dc32a..27e521a1 100644 --- a/src/conductor/ai/agents/runtime/_worker_entries.py +++ b/src/conductor/ai/agents/runtime/_worker_entries.py @@ -29,11 +29,14 @@ (``TypeError: isinstance() arg 2 must be a type``). """ +import hashlib import importlib import inspect +import marshal import multiprocessing import pickle import sys +from collections import deque from dataclasses import dataclass from typing import Any, Callable, Dict, FrozenSet, Optional @@ -67,78 +70,118 @@ def _walk_qualname(module_obj, qualname: str): # .coroutine; our Guardrail / ToolDef → .func. _CONTAINER_ATTRS = ("func", "coroutine") -def _extract_from_closure(func: Callable) -> Optional[Callable]: - """Extract the original user function from a closure's cell variables. +# How many attribute-nesting levels embedded-function discovery descends. +# openai-agents' FunctionTool -> on_invoke_tool (invoker instance) -> +# closure wrapper is two object edges deep. +_DEEP_EXTRACT_MAX_DEPTH = 2 + - - Shared by :func:`_find_embedded_function` below. - - In turn shared by :mod:`conductor.ai.agents.frameworks.serializer`'s - discovery and :class:`FunctionRef`'s parent-verification / - child-reconstruction — one implementation, can't drift apart. +def _is_embedded_function_candidate(value: Any) -> bool: + """Whether *value* is safely reconstructible as a function reference.""" + if not inspect.isfunction(value): + return False + module = getattr(value, "__module__", None) + qualname = getattr(value, "__qualname__", None) + return bool( + isinstance(module, str) + and isinstance(qualname, str) + and "" not in qualname + and "" not in qualname + ) + + +def _is_embedded_leaf(value: Any) -> bool: + """Return whether *value* cannot contain a useful embedded callable.""" + return ( + value is None + or isinstance(value, (bool, int, float, complex, str, bytes, bytearray)) + or inspect.ismodule(value) + or isinstance(value, type) + ) + + +def _iter_embedded_functions( + root: Any, max_depth: int = _DEEP_EXTRACT_MAX_DEPTH +): + """Yield embedded plain functions in deterministic breadth-first order. + + Container-held callables are common in framework decorators. This walk is + deliberately shape-based: it reads only instance dictionaries, never + invokes descriptors, and examines closure cells in their stored order. + ``serializer.py`` imports the compatibility helper below, so parent-side + discovery and child-side resolution share this exact traversal. """ - closure = getattr(func, "__closure__", None) - if not closure: - return None + queue = deque([(root, 0)]) + visited_object_ids = set() + yielded_function_ids = set() + + while queue: + obj, attribute_depth = queue.popleft() + object_id = id(obj) + if object_id in visited_object_ids: + continue + visited_object_ids.add(object_id) + + if inspect.isfunction(obj): + closure = getattr(obj, "__closure__", None) + if closure: + for cell in closure: + try: + value = cell.cell_contents + except ValueError: + continue + if _is_embedded_function_candidate(value): + value_id = id(value) + if value_id not in yielded_function_ids: + yielded_function_ids.add(value_id) + yield value + if _is_embedded_function_candidate(obj): + if object_id not in yielded_function_ids: + yielded_function_ids.add(object_id) + yield obj + continue - for cell in closure: + if attribute_depth >= max_depth or _is_embedded_leaf(obj): + continue try: - val = cell.cell_contents - except ValueError: + attributes = vars(obj) + except Exception: continue - if inspect.isfunction(val): - # Skip internal wrappers that take (ctx, input) or (context, ...) - try: - sig = inspect.signature(val) - param_names = list(sig.parameters.keys()) - # Internal wrappers typically start with ctx/context as first param - if param_names and param_names[0] in ("ctx", "context"): - continue - return val - except (ValueError, TypeError): - continue - return None + for name in sorted(attributes): + value = attributes[name] + if not _is_embedded_leaf(value): + queue.append((value, attribute_depth + 1)) -# How many attribute-nesting levels _find_embedded_function will descend. -# openai-agents' FunctionTool -> on_invoke_tool (invoker instance) -> -# _invoke_tool_impl (closure fn) is 2 levels deep; matches serializer.py's -# own default so parent discovery and child reconstruction stay in lockstep. -_DEEP_EXTRACT_MAX_DEPTH = 2 - +def _find_embedded_function( + obj: Any, max_depth: int = _DEEP_EXTRACT_MAX_DEPTH +) -> Optional[Callable]: + """Return the first deterministic embedded-function candidate, if any. -def _find_embedded_function(obj: Any, max_depth: int = _DEEP_EXTRACT_MAX_DEPTH) -> Optional[Callable]: - """Walk an object's attributes to find an embedded plain function. - - - Generic, framework-agnostic: no hardcoded attribute names. - - For containers that expose the original function only behind further - nested attributes and/or closures, not a single named attribute — e.g. - openai-agents' ``FunctionTool.on_invoke_tool`` is itself a callable - *instance*, not a function; the real closure lives one level deeper, - on that instance's own ``_invoke_tool_impl`` attribute. - - Deliberately avoids hardcoding that private attribute name: a - shape-based walk degrades to "not found" (the pre-existing, actionable - ``SpawnSafetyError``) instead of breaking outright if openai-agents - restructures its internals. + Kept as the serializer-facing compatibility API. FunctionRef itself never + relies on this convenience selection: it records a key for the exact + parent candidate and proves the child resolves that same candidate. """ - if max_depth <= 0: - return None + return next(_iter_embedded_functions(obj, max_depth), None) - for attr_name in vars(obj) if hasattr(obj, "__dict__") else []: - val = getattr(obj, attr_name, None) - if val is None: - continue - if inspect.isfunction(val): - func = _extract_from_closure(val) - if func is not None: - return func - return val +@dataclass(frozen=True) +class FunctionKey: + """Stable identity for a function across parent and spawn-child imports.""" - if hasattr(val, "__dict__") and not isinstance(val, type): - result = _find_embedded_function(val, max_depth - 1) - if result is not None: - return result + module: str + qualname: str + code_sha256: str - return None + @classmethod + def of(cls, fn: Callable) -> "FunctionKey": + if not inspect.isfunction(fn): + raise SpawnSafetyError(f"{fn!r} is not a plain function") + return cls( + fn.__module__, + fn.__qualname__, + hashlib.sha256(marshal.dumps(fn.__code__)).hexdigest(), + ) def _wrapped_depth_to(obj, fn) -> Optional[int]: @@ -189,6 +232,16 @@ class FunctionRef: unwrap_depth: int = 0 attr_hop: str = "" deep_extract: bool = False + expected_key: Optional[FunctionKey] = None + + def __post_init__(self): + if self.deep_extract: + if self.attr_hop or self.unwrap_depth or self.expected_key is None: + raise ValueError( + "deep extraction requires expected_key with no attr_hop or unwrap_depth" + ) + elif self.expected_key is not None: + raise ValueError("expected_key is only valid for deep extraction") @classmethod def of(cls, fn: Callable) -> "FunctionRef": @@ -253,13 +306,18 @@ def of(cls, fn: Callable) -> "FunctionRef": # always arrives as _find_embedded_function's own prior return value # (how serializer.py derives WorkerInfo.func) — identical walk always # matches by identity. - if _find_embedded_function(obj) is fn: - return cls(module, qualname, 0, "", deep_extract=True) + if any(candidate is fn for candidate in _iter_embedded_functions(obj)): + return cls( + module, + qualname, + deep_extract=True, + expected_key=FunctionKey.of(fn), + ) raise SpawnSafetyError( - f"'{module}.{qualname}' does not resolve back to {fn!r} (rebound " - f"without a __wrapped__ chain, a func/coroutine container " - f"attribute, or a discoverable nested/closure-held function). " - f"{_REMEDIES}" + f"'{module}.{qualname}' does not resolve back to the requested " + f"container-held callable {fn!r}: the rebound container has no " + f"__wrapped__ chain, func/coroutine attribute, or deterministic " + f"nested/closure discovery. {_REMEDIES}" ) def resolve(self) -> Callable: @@ -267,19 +325,46 @@ def resolve(self) -> Callable: cached = _RESOLVE_CACHE.get(self) if cached is not None: return cached - module_obj = importlib.import_module(self.module) - obj = _walk_qualname(module_obj, self.qualname) - if self.attr_hop: - obj = getattr(obj, self.attr_hop) + try: + module_obj = importlib.import_module(self.module) + obj = _walk_qualname(module_obj, self.qualname) + if self.attr_hop: + obj = getattr(obj, self.attr_hop) + except (AttributeError, ImportError) as exc: + raise SpawnSafetyError( + f"cannot resolve '{self.module}.{self.qualname}' using its " + f"recorded {'container attribute' if self.attr_hop else 'public name'} " + f"strategy: {exc}" + ) from None if self.deep_extract: - obj = _find_embedded_function(obj) - if obj is None: + matches = [] + seen_match_ids = set() + for candidate in _iter_embedded_functions(obj): + if FunctionKey.of(candidate) == self.expected_key: + candidate_id = id(candidate) + if candidate_id not in seen_match_ids: + seen_match_ids.add(candidate_id) + matches.append(candidate) + if len(matches) != 1: + outcome = "no matching callable" if not matches else "multiple matching callables" raise SpawnSafetyError( - f"'{self.module}.{self.qualname}' no longer yields a " - f"discoverable nested function (definition changed?)." + f"'{self.module}.{self.qualname}' deep extraction found " + f"{outcome} for {self.expected_key}; the container-held " + f"callable is inaccessible or ambiguous after import." ) - for _ in range(self.unwrap_depth): - obj = obj.__wrapped__ + obj = matches[0] + try: + for _ in range(self.unwrap_depth): + obj = obj.__wrapped__ + except AttributeError as exc: + raise SpawnSafetyError( + f"'{self.module}.{self.qualname}' cannot follow its recorded " + f"__wrapped__ strategy: {exc}" + ) from None + if not inspect.isfunction(obj): + raise SpawnSafetyError( + f"'{self.module}.{self.qualname}' resolved to {obj!r}, not a plain function." + ) _RESOLVE_CACHE[self] = obj return obj @@ -315,18 +400,17 @@ def __init__(self, tool_name, fn_ref=None, fn_direct=None, guardrails=None, def for_callable(cls, fn, tool_name, guardrails=None, credential_names=None): """Build an entry for *fn*, preferring by-reference transport. - Falls back to direct transport for picklable callables (instances, - bound methods of picklable objects); unpicklable closures are caught - by the registration probe with an actionable error. + Plain functions must resolve by reference so a decorator-container + mismatch is reported during registration. Other callable objects keep + the legacy direct transport and are validated by the spawn probe. """ framework = bool(getattr(fn, "_conductor_agent_framework_callable", False)) - try: + if inspect.isfunction(fn): ref = FunctionRef.of(fn) - except SpawnSafetyError: - entry = cls(tool_name, fn_direct=fn, guardrails=guardrails, + entry = cls(tool_name, fn_ref=ref, guardrails=guardrails, credential_names=credential_names, framework_callable=framework) else: - entry = cls(tool_name, fn_ref=ref, guardrails=guardrails, + entry = cls(tool_name, fn_direct=fn, guardrails=guardrails, credential_names=credential_names, framework_callable=framework) # Introspection compatibility (logging etc.). Plain string INSTANCE # attrs — unlike the old wrapper's reassigned function identity, these From 1a0a27be0e3f0a940dd9368bbbc58865aedd9b8e Mon Sep 17 00:00:00 2001 From: Viren Baraiya Date: Sun, 9 Aug 2026 13:43:45 -0700 Subject: [PATCH 6/9] test_cycle repair --- e2e/test_suite4_mcp_tools.py | 5 +++++ e2e/test_suite5_http_tools.py | 5 +++++ 2 files changed, 10 insertions(+) diff --git a/e2e/test_suite4_mcp_tools.py b/e2e/test_suite4_mcp_tools.py index 4bcdfd51..f49e8059 100644 --- a/e2e/test_suite4_mcp_tools.py +++ b/e2e/test_suite4_mcp_tools.py @@ -21,6 +21,11 @@ from conductor.ai.agents import Agent, mcp_tool +pytest.importorskip( + "mcp_test_server", + reason="mcp-testkit Python package is required for Suite 4 MCP tools test", +) + pytestmark = [ pytest.mark.e2e, pytest.mark.xdist_group("credentials"), diff --git a/e2e/test_suite5_http_tools.py b/e2e/test_suite5_http_tools.py index cbabe799..4321c728 100644 --- a/e2e/test_suite5_http_tools.py +++ b/e2e/test_suite5_http_tools.py @@ -21,6 +21,11 @@ from conductor.ai.agents import Agent, api_tool, http_tool +pytest.importorskip( + "mcp_test_server", + reason="mcp-testkit Python package is required for Suite 5 HTTP tools test", +) + pytestmark = [ pytest.mark.e2e, pytest.mark.xdist_group("credentials"), From 9b1cc9c2fe10d748d1b8dbe045be301ec4bd8069 Mon Sep 17 00:00:00 2001 From: Viren Baraiya Date: Mon, 10 Aug 2026 00:57:12 -0700 Subject: [PATCH 7/9] code_subtask fix-openai-spawn-test --- tests/unit/ai/test_worker_entries.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/ai/test_worker_entries.py b/tests/unit/ai/test_worker_entries.py index f21a9927..09a74c74 100644 --- a/tests/unit/ai/test_worker_entries.py +++ b/tests/unit/ai/test_worker_entries.py @@ -214,7 +214,7 @@ def test_entry_transports_function_tool_fn_by_ref(self, is_async): assert clone._target() is raw def test_sync_function_tool_entry_executes_in_spawn_child(self): - _, raw, _ = self._openai_tools() + oa, raw, _ = self._openai_tools() assert raw is not None entry_bytes = pickle.dumps(ToolWorkerEntry.for_callable(raw, "oa_get_weather")) @@ -234,7 +234,7 @@ def test_sync_function_tool_entry_executes_in_spawn_child(self): assert output == {"result": "sunny in Boston"} def test_async_function_tool_ref_executes_in_spawn_child(self): - _, _, raw = self._openai_tools() + oa, _, raw = self._openai_tools() assert raw is not None ctx = multiprocessing.get_context("spawn") q = ctx.Queue() From e8ac2c34702cd34d04170e610277f5e3b7b4afbe Mon Sep 17 00:00:00 2001 From: Viren Baraiya Date: Mon, 10 Aug 2026 00:59:16 -0700 Subject: [PATCH 8/9] code_subtask retry-signal-lock-races --- tests/integration/retry_helpers.py | 30 ++++---- .../workflow/test_workflow_execution.py | 69 +++++++++++++++---- 2 files changed, 75 insertions(+), 24 deletions(-) diff --git a/tests/integration/retry_helpers.py b/tests/integration/retry_helpers.py index 4f69ed79..b37a868c 100644 --- a/tests/integration/retry_helpers.py +++ b/tests/integration/retry_helpers.py @@ -66,10 +66,11 @@ def is_transient(exc): or exc.status in GATEWAY_STATUSES) -def first_transient_api_exception(exc): +def first_transient_api_exception(exc, retry_statuses=()): """Walk the exception chain (``__cause__`` / ``__context__``) and return the first transient ``ApiException`` (flagged transient, or status 0/None), or - ``None`` if there isn't one. + one whose status is explicitly listed in ``retry_statuses``. Return ``None`` + when there isn't one. Inner test helpers sometimes catch an ApiException and re-raise it as a bare ``Exception`` (losing the type), so we can't rely on the outermost exception @@ -80,7 +81,8 @@ def first_transient_api_exception(exc): cur = exc while cur is not None and id(cur) not in seen: seen.add(id(cur)) - if is_transient(cur): + if is_transient(cur) or ( + isinstance(cur, ApiException) and cur.status in retry_statuses): return cur cur = cur.__cause__ or cur.__context__ return None @@ -139,10 +141,12 @@ def retry_on_status(func, *args, statuses=(404,), retries=5, base_delay=1.0, def retry_scenario(label, func, *args, deadline=None, base_delay=DEFAULT_BASE_DELAY_SECONDS, - max_delay=DEFAULT_MAX_DELAY_SECONDS, **kwargs): + max_delay=DEFAULT_MAX_DELAY_SECONDS, retry_statuses=(), + **kwargs): """Run ``func(*args, **kwargs)``, retrying only on a transient blip (see ``is_transient``: status 0/None transport hiccups plus gateway-class - 502/503/504) until the shared ``deadline`` passes. + 502/503/504), plus statuses explicitly listed in ``retry_statuses``, until + the shared ``deadline`` passes. Args: label: Human-readable scenario name for logs. @@ -151,6 +155,8 @@ def retry_scenario(label, func, *args, deadline=None, ``None`` (e.g. the standalone ``main.py`` runner), the scenario runs exactly once with no transient retry — preserving prior behaviour. base_delay / max_delay: capped exponential backoff bounds (seconds). + retry_statuses: additional HTTP statuses that this scenario explicitly + opts into retrying. Defaults to no additional statuses. Non-transient errors (assertion failures, genuine 4xx/5xx) raise immediately. A transient blip at/after the deadline re-raises so the suite @@ -166,23 +172,23 @@ def retry_scenario(label, func, *args, deadline=None, try: return func(*args, **kwargs) except Exception as e: - transient = first_transient_api_exception(e) - if transient is None: + retryable = first_transient_api_exception(e, retry_statuses) + if retryable is None: raise now = time.monotonic() if now >= deadline: logger.error( - 'transient (%s) in %s but overall deadline exceeded by ' + 'retryable (%s) in %s but overall deadline exceeded by ' '%.0fs; giving up: %s', - transient.status, label, now - deadline, transient) + retryable.status, label, now - deadline, retryable) raise delay = min(base_delay * (2 ** attempt), max_delay) delay = min(delay, max(0.0, deadline - now)) logger.warning( - 'transient (%s) in %s (attempt %d); retrying in %.1fs ' + 'retryable (%s) in %s (attempt %d); retrying in %.1fs ' '(%.0fs left in overall budget): %s', - transient.status, label, attempt + 1, delay, - deadline - now, transient) + retryable.status, label, attempt + 1, delay, + deadline - now, retryable) time.sleep(delay) attempt += 1 diff --git a/tests/integration/workflow/test_workflow_execution.py b/tests/integration/workflow/test_workflow_execution.py index 49e534bf..118c60a0 100644 --- a/tests/integration/workflow/test_workflow_execution.py +++ b/tests/integration/workflow/test_workflow_execution.py @@ -6,6 +6,7 @@ from conductor.client.configuration.configuration import Configuration from conductor.client.http.models import StartWorkflowRequest from conductor.client.http.models import TaskDef +from conductor.client.http.rest import ApiException from conductor.client.worker.worker import ExecuteTaskFunction from conductor.client.worker.worker import Worker from conductor.client.workflow.conductor_workflow import ConductorWorkflow @@ -462,14 +463,51 @@ def _wait_for_workflow_completion(workflow_executor: WorkflowExecutor, workflow_ # ===== SIGNAL TESTS ===== +SIGNAL_RETRY_STATUSES = {423} + + +def test_retry_scenario_retries_explicit_statuses_only(): + attempts = [] + + def raise_locked(): + attempts.append('attempt') + raise ApiException(status=423, reason='Locked') + + try: + retry_scenario( + 'locked_without_opt_in', raise_locked, + deadline=time.monotonic() + 1, base_delay=0, max_delay=0) + assert False, 'HTTP 423 should not retry without an explicit opt-in' + except ApiException as exc: + assert exc.status == 423 + + assert attempts == ['attempt'] + + attempts.clear() + + def complete_on_second_attempt(): + attempts.append('attempt') + if len(attempts) == 1: + raise ApiException(status=423, reason='Locked') + return 'fresh-workflow' + + result = retry_scenario( + 'locked_with_opt_in', complete_on_second_attempt, + deadline=time.monotonic() + 1, base_delay=0, max_delay=0, + retry_statuses={423}) + + assert result == 'fresh-workflow' + assert attempts == ['attempt', 'attempt'] + + def run_signal_tests(configuration: Configuration, workflow_executor: WorkflowExecutor, deadline=None): """Run all signal API tests using WorkflowExecutor methods. - Each scenario is retried at the scenario level on a transient blip (see - retry_scenario): a retry starts a fresh workflow and issues a fresh sync - signal, so the asserted SignalResponse is always from a signal this attempt - actually sent — no double-signalling of a single workflow. + Each scenario is retried at the scenario level on a transient blip or HTTP + 423 (see retry_scenario): a retry starts a fresh workflow and issues a fresh + sync signal, so the asserted SignalResponse is always from a signal this + attempt actually sent — no double-signalling of a single workflow. """ logger.info('START: Signal API tests using WorkflowExecutor') @@ -481,25 +519,32 @@ def run_signal_tests(configuration: Configuration, workflow_executor: WorkflowEx # Test sync signal with different return strategies retry_scenario('scenario_signal_target_workflow', - scenario_signal_target_workflow, workflow_executor, deadline=deadline) + scenario_signal_target_workflow, workflow_executor, + deadline=deadline, retry_statuses=SIGNAL_RETRY_STATUSES) retry_scenario('scenario_signal_blocking_workflow', - scenario_signal_blocking_workflow, workflow_executor, deadline=deadline) + scenario_signal_blocking_workflow, workflow_executor, + deadline=deadline, retry_statuses=SIGNAL_RETRY_STATUSES) retry_scenario('scenario_signal_blocking_task', - scenario_signal_blocking_task, workflow_executor, deadline=deadline) + scenario_signal_blocking_task, workflow_executor, + deadline=deadline, retry_statuses=SIGNAL_RETRY_STATUSES) retry_scenario('scenario_signal_blocking_task_input', - scenario_signal_blocking_task_input, workflow_executor, deadline=deadline) + scenario_signal_blocking_task_input, workflow_executor, + deadline=deadline, retry_statuses=SIGNAL_RETRY_STATUSES) # Test default return strategy retry_scenario('scenario_signal_default_strategy', - scenario_signal_default_strategy, workflow_executor, deadline=deadline) + scenario_signal_default_strategy, workflow_executor, + deadline=deadline, retry_statuses=SIGNAL_RETRY_STATUSES) # Test async signal retry_scenario('scenario_signal_async', - scenario_signal_async, workflow_executor, deadline=deadline) + scenario_signal_async, workflow_executor, + deadline=deadline, retry_statuses=SIGNAL_RETRY_STATUSES) # Test to_dict fix retry_scenario('scenario_signal_to_dict_fix', - scenario_signal_to_dict_fix, workflow_executor, deadline=deadline) + scenario_signal_to_dict_fix, workflow_executor, + deadline=deadline, retry_statuses=SIGNAL_RETRY_STATUSES) logger.info('All signal tests completed successfully') @@ -863,4 +908,4 @@ def scenario_signal_to_dict_fix(workflow_executor: WorkflowExecutor): _wait_for_workflow_completion(workflow_executor, workflow_id) - logger.info('to_dict() method test completed') \ No newline at end of file + logger.info('to_dict() method test completed') From 57d7fc6c882a50796d53e1bd86d91ac3a6b1709a Mon Sep 17 00:00:00 2001 From: Viren Baraiya Date: Mon, 10 Aug 2026 01:00:31 -0700 Subject: [PATCH 9/9] test_cycle: agent-authored test for previously unmapped changes --- tests/integration/test_retry_helpers.py | 31 +++++++++++++++++++++++++ 1 file changed, 31 insertions(+) create mode 100644 tests/integration/test_retry_helpers.py diff --git a/tests/integration/test_retry_helpers.py b/tests/integration/test_retry_helpers.py new file mode 100644 index 00000000..68e3a79b --- /dev/null +++ b/tests/integration/test_retry_helpers.py @@ -0,0 +1,31 @@ +import unittest +from unittest.mock import patch + +from conductor.client.http.rest import ApiException +from tests.integration.retry_helpers import retry_scenario + + +class TestRetryScenario(unittest.TestCase): + + @patch('tests.integration.retry_helpers.time.sleep') + @patch('tests.integration.retry_helpers.time.monotonic', return_value=100.0) + def test_retries_wrapped_explicit_status(self, monotonic, sleep): + attempts = [] + + def signal_fresh_workflow(): + attempts.append('attempt') + if len(attempts) == 1: + try: + raise ApiException(status=423, reason='Locked') + except ApiException as exc: + raise RuntimeError('signal race') from exc + return 'fresh-workflow' + + result = retry_scenario( + 'signal_race', signal_fresh_workflow, deadline=200.0, + base_delay=0.25, max_delay=1.0, retry_statuses=(423,)) + + self.assertEqual(result, 'fresh-workflow') + self.assertEqual(attempts, ['attempt', 'attempt']) + monotonic.assert_called_once_with() + sleep.assert_called_once_with(0.25)