From 890d891934eca7a03e25db310dce023b28aab89c Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Mon, 10 Aug 2026 16:35:49 -0400
Subject: [PATCH 01/15] Resolve job inputs in decorated flows

---
 src/jobflow/core/job.py           | 21 +++++++++++++++++++++
 tests/core/test_flow_decorator.py | 20 ++++++++++++++++++++
 2 files changed, 41 insertions(+)

diff --git a/src/jobflow/core/job.py b/src/jobflow/core/job.py
index d00501b2..b08bcef6 100644
--- a/src/jobflow/core/job.py
+++ b/src/jobflow/core/job.py
@@ -28,6 +28,23 @@
 logger = logging.getLogger(__name__)
 
 
+def _replace_job_or_flow_with_output(value):
+    """Replace Jobs and Flows in nested containers with their outputs."""
+    from jobflow.core.flow import Flow
+
+    if isinstance(value, (Job, Flow)):
+        return value.output
+    if isinstance(value, list):
+        return [_replace_job_or_flow_with_output(item) for item in value]
+    if isinstance(value, tuple):
+        return tuple(_replace_job_or_flow_with_output(item) for item in value)
+    if isinstance(value, dict):
+        return {
+            key: _replace_job_or_flow_with_output(item) for key, item in value.items()
+        }
+    return value
+
+
 @dataclass
 class JobConfig(MSONable):
     """
@@ -213,6 +230,10 @@ def get_job(*args, **kwargs) -> Job:
                         f = met
                         args = args[1:]
 
+            if _current_flow_context.get() is not None:
+                args = _replace_job_or_flow_with_output(args)
+                kwargs = _replace_job_or_flow_with_output(kwargs)
+
             return Job(
                 function=f, function_args=args, function_kwargs=kwargs, **job_kwargs
             )
diff --git a/tests/core/test_flow_decorator.py b/tests/core/test_flow_decorator.py
index aa6b755d..58d8b099 100644
--- a/tests/core/test_flow_decorator.py
+++ b/tests/core/test_flow_decorator.py
@@ -159,6 +159,26 @@ def my_flow(a, b):
     assert result[flow1.output.uuid][1].output == 7
 
 
+def test_flow_resolves_job_inputs_to_outputs():
+    """Test that Jobs used as inputs inside a decorated flow resolve to outputs."""
+    from jobflow import flow, job
+    from jobflow.managers.local import run_locally
+
+    @job
+    def combine(a, values):
+        return a + values["nested"][0]
+
+    @flow
+    def my_flow(a, b):
+        sum_job = add(a, b)
+        return combine(a, {"nested": [sum_job]})
+
+    flow1 = my_flow(1, 2)
+    result = run_locally(flow1, ensure_success=True)
+
+    assert result[flow1.output.uuid][1].output == 4
+
+
 def test_flow_returns_list():
     """Test that a flow that returns a list of OutputReferences
     can be created and run."""

From 9f186cc523d3df0eef55b549e264ab97a97da647 Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Mon, 10 Aug 2026 17:21:16 -0400
Subject: [PATCH 02/15] Fix VCS fallback version

---
 pyproject.toml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/pyproject.toml b/pyproject.toml
index 6e8d3714..17a79973 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -79,7 +79,7 @@ jobflow = ["py.typed"]
 
 [tool.versioningit.vcs]
 method = "git"
-default-tag = "0.0.1"
+default-tag = "0.3.1"
 
 [tool.flake8]
 max-line-length = 88

From 8cc03209f69efe7629136ad424802fe0fcb36fad Mon Sep 17 00:00:00 2001
From: Andrew-S-Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 10:18:28 -0400
Subject: [PATCH 03/15] Retain jobs created by dynamic subflows

---
 src/jobflow/core/flow.py          | 18 ++++++++++++++-
 src/jobflow/core/job.py           | 38 +++++++++++++------------------
 tests/core/test_flow_decorator.py | 13 +++++++++++
 tests/core/test_job.py            | 21 ++++++++++++++++-
 4 files changed, 66 insertions(+), 24 deletions(-)

diff --git a/src/jobflow/core/flow.py b/src/jobflow/core/flow.py
index 17fb2fd7..8e58af7a 100644
--- a/src/jobflow/core/flow.py
+++ b/src/jobflow/core/flow.py
@@ -931,6 +931,22 @@ def get_flow(
     return flow
 
 
+def _normalize_job_or_flow(value):
+    """Replace Jobs and Flows in nested containers with their outputs."""
+    if isinstance(value, (jobflow.Job, jobflow.Flow)):
+        return value.output
+    if isinstance(value, list):
+        return [_normalize_job_or_flow(item) for item in value]
+    if isinstance(value, tuple):
+        return tuple(_normalize_job_or_flow(item) for item in value)
+    if isinstance(value, dict):
+        return {
+            _normalize_job_or_flow(key): _normalize_job_or_flow(item)
+            for key, item in value.items()
+        }
+    return value
+
+
 class DecoratedFlow(Flow):
     """A DecoratedFlow is a Flow that is returned on using the @flow decorator."""
 
@@ -973,7 +989,7 @@ def __init__(self, fn, *args, **kwargs):
                 f"of your @flow decorated function.",
                 stacklevel=2,
             )
-            output = output.output
+        output = _normalize_job_or_flow(output)
 
         super().__init__(name=name, jobs=children_list, output=output)
 
diff --git a/src/jobflow/core/job.py b/src/jobflow/core/job.py
index b08bcef6..c0b8f032 100644
--- a/src/jobflow/core/job.py
+++ b/src/jobflow/core/job.py
@@ -11,7 +11,7 @@
 from monty.json import MSONable, jsanitize
 from typing_extensions import Self
 
-from jobflow.core.flow import _current_flow_context
+from jobflow.core.flow import _current_flow_context, _normalize_job_or_flow
 from jobflow.core.reference import OnMissing, OutputReference
 from jobflow.utils.uid import suid
 
@@ -28,23 +28,6 @@
 logger = logging.getLogger(__name__)
 
 
-def _replace_job_or_flow_with_output(value):
-    """Replace Jobs and Flows in nested containers with their outputs."""
-    from jobflow.core.flow import Flow
-
-    if isinstance(value, (Job, Flow)):
-        return value.output
-    if isinstance(value, list):
-        return [_replace_job_or_flow_with_output(item) for item in value]
-    if isinstance(value, tuple):
-        return tuple(_replace_job_or_flow_with_output(item) for item in value)
-    if isinstance(value, dict):
-        return {
-            key: _replace_job_or_flow_with_output(item) for key, item in value.items()
-        }
-    return value
-
-
 @dataclass
 class JobConfig(MSONable):
     """
@@ -231,8 +214,8 @@ def get_job(*args, **kwargs) -> Job:
                         args = args[1:]
 
             if _current_flow_context.get() is not None:
-                args = _replace_job_or_flow_with_output(args)
-                kwargs = _replace_job_or_flow_with_output(kwargs)
+                args = _normalize_job_or_flow(args)
+                kwargs = _normalize_job_or_flow(kwargs)
 
             return Job(
                 function=f, function_args=args, function_kwargs=kwargs, **job_kwargs
@@ -625,7 +608,12 @@ def run(self, store: jobflow.JobStore, job_dir: Path = None) -> Response:
         from datetime import datetime
 
         from jobflow import CURRENT_JOB
-        from jobflow.core.flow import get_flow
+        from jobflow.core.flow import (
+            Flow,
+            _normalize_job_or_flow,
+            flow_build_context,
+            get_flow,
+        )
         from jobflow.core.schemas import JobStoreDocument
 
         index_str = f", {self.index}" if self.index != 1 else ""
@@ -647,7 +635,13 @@ def run(self, store: jobflow.JobStore, job_dir: Path = None) -> Response:
         if bound is not None and not isinstance(bound, types.ModuleType):
             function = types.MethodType(function, bound)
 
-        response = function(*self.function_args, **self.function_kwargs)
+        children = []
+        with flow_build_context(children):
+            response = function(*self.function_args, **self.function_kwargs)
+
+        children = [child for child in children if child.host is None]
+        if children and not isinstance(response, Response):
+            response = Flow(jobs=children, output=_normalize_job_or_flow(response))
         response = Response.from_job_returns(
             response, self.output_schema, job_dir=job_dir
         )
diff --git a/tests/core/test_flow_decorator.py b/tests/core/test_flow_decorator.py
index 58d8b099..2290b82f 100644
--- a/tests/core/test_flow_decorator.py
+++ b/tests/core/test_flow_decorator.py
@@ -179,6 +179,19 @@ def my_flow(a, b):
     assert result[flow1.output.uuid][1].output == 4
 
 
+def test_flow_normalizes_nested_job_outputs():
+    """Test Jobs nested in decorated flow outputs resolve to references."""
+    from jobflow import flow
+
+    @flow
+    def my_flow():
+        return {"nested": (add(1, 2),)}
+
+    flow1 = my_flow()
+
+    assert flow1.output == {"nested": (flow1.jobs[0].output,)}
+
+
 def test_flow_returns_list():
     """Test that a flow that returns a list of OutputReferences
     can be created and run."""
diff --git a/tests/core/test_job.py b/tests/core/test_job.py
index f9416221..3bb73c8c 100644
--- a/tests/core/test_job.py
+++ b/tests/core/test_job.py
@@ -83,7 +83,6 @@ def test_job_run(capsys, memory_jobstore, memory_data_jobstore):
     response = test_job.run(memory_jobstore)
     assert capsys.readouterr().out == "I am a job\n"
     assert isinstance(response, Response)
-
     # test run with outputs
     test_job = Job(add, function_args=(1,), function_kwargs={"b": 2})
     response = test_job.run(memory_jobstore)
@@ -1447,3 +1446,23 @@ def add_configured(a, b):
         f"Expected job2.config.manager_config to be {{'key': 'original'}}, "
         f"but got {job2.config.manager_config!r} — shared instance bug confirmed."
     )
+
+
+def test_job_collects_dynamic_subflow(memory_jobstore):
+    """Test a job retains every job created by a returned dynamic subflow."""
+    from jobflow import Flow, job
+
+    @job
+    def add_job(a, b):
+        return a + b
+
+    @job
+    def dynamic_subflow():
+        first = add_job(1, 2)
+        return add_job(first.output, 3)
+
+    response = dynamic_subflow().run(memory_jobstore)
+
+    assert isinstance(response.replace, Flow)
+    assert len(response.replace) == 3
+    assert response.replace.output.uuid == response.replace[-2].uuid

From 81075c11525825340f4010bab80fd964697ef7bb Mon Sep 17 00:00:00 2001
From: Andrew-S-Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 10:30:10 -0400
Subject: [PATCH 04/15] Format consolidated Jobflow changes

---
 tests/core/test_job.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/core/test_job.py b/tests/core/test_job.py
index 3bb73c8c..2236b722 100644
--- a/tests/core/test_job.py
+++ b/tests/core/test_job.py
@@ -1154,7 +1154,7 @@ def make(self):
     test_job = maker.make()
     test_job.update_metadata(
         {"v": 5},
-        callback_filter=lambda job: (job.maker is not None and job.maker.value == 42),
+        callback_filter=lambda job: job.maker is not None and job.maker.value == 42,
     )
     assert test_job.metadata["v"] == 5
 

From f2e02041ae3a854755680a70b03d5f1af9736573 Mon Sep 17 00:00:00 2001
From: Andrew-S-Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 10:32:52 -0400
Subject: [PATCH 05/15] Annotate dynamic flow children

---
 src/jobflow/core/job.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/jobflow/core/job.py b/src/jobflow/core/job.py
index c0b8f032..94052125 100644
--- a/src/jobflow/core/job.py
+++ b/src/jobflow/core/job.py
@@ -635,7 +635,7 @@ def run(self, store: jobflow.JobStore, job_dir: Path = None) -> Response:
         if bound is not None and not isinstance(bound, types.ModuleType):
             function = types.MethodType(function, bound)
 
-        children = []
+        children: list[Job | Flow] = []
         with flow_build_context(children):
             response = function(*self.function_args, **self.function_kwargs)
 

From bc8d8308acd087e180eb39a4d70ef5939aa6cc2a Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 12:41:28 -0400
Subject: [PATCH 06/15] test: cover minimal job input example

---
 tests/core/test_flow_decorator.py | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/tests/core/test_flow_decorator.py b/tests/core/test_flow_decorator.py
index 2290b82f..76f25750 100644
--- a/tests/core/test_flow_decorator.py
+++ b/tests/core/test_flow_decorator.py
@@ -159,6 +159,27 @@ def my_flow(a, b):
     assert result[flow1.output.uuid][1].output == 7
 
 
+def test_flow_resolves_minimal_job_input_example():
+    """Test the minimal Job-as-input example from issue #884."""
+    from jobflow import flow, job
+    from jobflow.managers.local import run_locally
+
+    @job
+    def mult(a, b):
+        return a * b
+
+    @flow
+    def workflow(a, b):
+        sum_result = add(a, b)
+        mult_result = mult(a, sum_result)
+        return mult_result
+
+    flow1 = workflow(1, 2)
+    result = run_locally(flow1, ensure_success=True)
+
+    assert result[flow1.output.uuid][1].output == 3
+
+
 def test_flow_resolves_job_inputs_to_outputs():
     """Test that Jobs used as inputs inside a decorated flow resolve to outputs."""
     from jobflow import flow, job

From 2b84e6544d8b2fee39065c6af8a8b78c3ae55c37 Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 12:43:37 -0400
Subject: [PATCH 07/15] test: clarify runtime job collection

---
 tests/core/test_job.py | 13 +++++++------
 1 file changed, 7 insertions(+), 6 deletions(-)

diff --git a/tests/core/test_job.py b/tests/core/test_job.py
index 2236b722..d1c41e0d 100644
--- a/tests/core/test_job.py
+++ b/tests/core/test_job.py
@@ -1448,21 +1448,22 @@ def add_configured(a, b):
     )
 
 
-def test_job_collects_dynamic_subflow(memory_jobstore):
-    """Test a job retains every job created by a returned dynamic subflow."""
-    from jobflow import Flow, job
+def test_job_collects_jobs_created_during_run(memory_jobstore):
+    """Test a job retains all jobs created while its function is running."""
+    from jobflow import job
 
     @job
     def add_job(a, b):
         return a + b
 
     @job
-    def dynamic_subflow():
+    def create_jobs():
         first = add_job(1, 2)
         return add_job(first.output, 3)
 
-    response = dynamic_subflow().run(memory_jobstore)
+    response = create_jobs().run(memory_jobstore)
 
-    assert isinstance(response.replace, Flow)
+    # The replacement contains both created jobs and the output-mapping job added by
+    # prepare_replace. Its output should refer to the second created job.
     assert len(response.replace) == 3
     assert response.replace.output.uuid == response.replace[-2].uuid

From 27ecf81649bc4abca51e887141f5bc704dd4b9fa Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 12:46:35 -0400
Subject: [PATCH 08/15] refactor: normalize decorated flow outputs once

---
 src/jobflow/core/flow.py          | 24 ++++++++++++------------
 tests/core/test_flow_decorator.py |  3 +--
 2 files changed, 13 insertions(+), 14 deletions(-)

diff --git a/src/jobflow/core/flow.py b/src/jobflow/core/flow.py
index 8e58af7a..5df0acf5 100644
--- a/src/jobflow/core/flow.py
+++ b/src/jobflow/core/flow.py
@@ -931,9 +931,19 @@ def get_flow(
     return flow
 
 
-def _normalize_job_or_flow(value):
+def _normalize_job_or_flow(value, decorated_flow_name=None):
     """Replace Jobs and Flows in nested containers with their outputs."""
     if isinstance(value, (jobflow.Job, jobflow.Flow)):
+        if decorated_flow_name is not None:
+            warnings.warn(
+                f"@flow decorated function '{decorated_flow_name}' contains a Flow "
+                "or Job as an output. Usually the output should be the output of "
+                "a Job or another Flow (e.g. job.output). Replacing the output of "
+                "the @flow with the output of the Flow/Job. If this message is "
+                "unexpected then double check the outputs of your @flow decorated "
+                "function.",
+                stacklevel=3,
+            )
         return value.output
     if isinstance(value, list):
         return [_normalize_job_or_flow(item) for item in value]
@@ -979,17 +989,7 @@ def __init__(self, fn, *args, **kwargs):
         ):
             name = args[0].name
 
-        if isinstance(output, (jobflow.Job, jobflow.Flow)):
-            warnings.warn(
-                f"@flow decorated function '{name}' contains a Flow or"
-                f"Job as an output. Usually the output should be the output of"
-                f"a Job or another Flow (e.g. job.output). Replacing the"
-                f"output of the @flow with the output of the Flow/Job."
-                f"If this message is unexpected then double check the outputs"
-                f"of your @flow decorated function.",
-                stacklevel=2,
-            )
-        output = _normalize_job_or_flow(output)
+        output = _normalize_job_or_flow(output, decorated_flow_name=name)
 
         super().__init__(name=name, jobs=children_list, output=output)
 
diff --git a/tests/core/test_flow_decorator.py b/tests/core/test_flow_decorator.py
index 76f25750..7dc4dc88 100644
--- a/tests/core/test_flow_decorator.py
+++ b/tests/core/test_flow_decorator.py
@@ -171,8 +171,7 @@ def mult(a, b):
     @flow
     def workflow(a, b):
         sum_result = add(a, b)
-        mult_result = mult(a, sum_result)
-        return mult_result
+        return mult(a, sum_result)
 
     flow1 = workflow(1, 2)
     result = run_locally(flow1, ensure_success=True)

From 713349120bf22205ac62b0d6f407c9016994e389 Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 12:48:11 -0400
Subject: [PATCH 09/15] refactor: keep decorated flow warning at call site

---
 src/jobflow/core/flow.py | 23 +++++++++++------------
 1 file changed, 11 insertions(+), 12 deletions(-)

diff --git a/src/jobflow/core/flow.py b/src/jobflow/core/flow.py
index 5df0acf5..82a61691 100644
--- a/src/jobflow/core/flow.py
+++ b/src/jobflow/core/flow.py
@@ -931,19 +931,9 @@ def get_flow(
     return flow
 
 
-def _normalize_job_or_flow(value, decorated_flow_name=None):
+def _normalize_job_or_flow(value):
     """Replace Jobs and Flows in nested containers with their outputs."""
     if isinstance(value, (jobflow.Job, jobflow.Flow)):
-        if decorated_flow_name is not None:
-            warnings.warn(
-                f"@flow decorated function '{decorated_flow_name}' contains a Flow "
-                "or Job as an output. Usually the output should be the output of "
-                "a Job or another Flow (e.g. job.output). Replacing the output of "
-                "the @flow with the output of the Flow/Job. If this message is "
-                "unexpected then double check the outputs of your @flow decorated "
-                "function.",
-                stacklevel=3,
-            )
         return value.output
     if isinstance(value, list):
         return [_normalize_job_or_flow(item) for item in value]
@@ -989,7 +979,16 @@ def __init__(self, fn, *args, **kwargs):
         ):
             name = args[0].name
 
-        output = _normalize_job_or_flow(output, decorated_flow_name=name)
+        if isinstance(output, (jobflow.Job, jobflow.Flow)):
+            warnings.warn(
+                f"@flow decorated function '{name}' contains a Flow or Job as an "
+                "output. Usually the output should be the output of a Job or another "
+                "Flow (e.g. job.output). Replacing the output of the @flow with the "
+                "output of the Flow/Job. If this message is unexpected then double "
+                "check the outputs of your @flow decorated function.",
+                stacklevel=2,
+            )
+        output = _normalize_job_or_flow(output)
 
         super().__init__(name=name, jobs=children_list, output=output)
 

From 6f20c0caf05343196e2142e7a3e74c5c386756dc Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 12:51:45 -0400
Subject: [PATCH 10/15] perf: avoid rechecking normalized flow outputs

---
 src/jobflow/core/flow.py          |  8 +++++++-
 tests/core/test_flow_decorator.py | 18 ++++++++++++++++++
 2 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/src/jobflow/core/flow.py b/src/jobflow/core/flow.py
index 82a61691..13193bdc 100644
--- a/src/jobflow/core/flow.py
+++ b/src/jobflow/core/flow.py
@@ -284,8 +284,11 @@ def output(self, output: Any):
             The output of the flow. These should come from the output of one
             or more of the jobs.
         """
+        output_is_normalized = getattr(self, "_output_is_normalized", False)
+        self._output_is_normalized = False
+
         if output is not None:
-            if contains_flow_or_job(output):
+            if not output_is_normalized and contains_flow_or_job(output):
                 warnings.warn(
                     f"Flow '{self.name}' contains a Flow or Job as an output. "
                     f"Usually the Flow output should be the output of a Job or "
@@ -989,6 +992,9 @@ def __init__(self, fn, *args, **kwargs):
                 stacklevel=2,
             )
         output = _normalize_job_or_flow(output)
+        # Avoid checking the entire output again in Flow.output; normalization has
+        # already replaced every nested Job and Flow.
+        self._output_is_normalized = True
 
         super().__init__(name=name, jobs=children_list, output=output)
 
diff --git a/tests/core/test_flow_decorator.py b/tests/core/test_flow_decorator.py
index 7dc4dc88..14747fd5 100644
--- a/tests/core/test_flow_decorator.py
+++ b/tests/core/test_flow_decorator.py
@@ -212,6 +212,24 @@ def my_flow():
     assert flow1.output == {"nested": (flow1.jobs[0].output,)}
 
 
+def test_flow_does_not_recheck_normalized_output(monkeypatch):
+    """Test normalized decorated-flow outputs are not scanned a second time."""
+    from jobflow import flow
+
+    def fail_if_called(_):
+        raise AssertionError("normalized output was checked again")
+
+    monkeypatch.setattr("jobflow.core.flow.contains_flow_or_job", fail_if_called)
+
+    @flow
+    def my_flow():
+        return {"nested": (add(1, 2),)}
+
+    flow1 = my_flow()
+
+    assert flow1.output == {"nested": (flow1.jobs[0].output,)}
+
+
 def test_flow_returns_list():
     """Test that a flow that returns a list of OutputReferences
     can be created and run."""

From e4b1f76ad45aba77ee7cb56aa8d27cd1601a568d Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 12:54:08 -0400
Subject: [PATCH 11/15] Revert "perf: avoid rechecking normalized flow outputs"

This reverts commit 6f20c0caf05343196e2142e7a3e74c5c386756dc.
---
 src/jobflow/core/flow.py          |  8 +-------
 tests/core/test_flow_decorator.py | 18 ------------------
 2 files changed, 1 insertion(+), 25 deletions(-)

diff --git a/src/jobflow/core/flow.py b/src/jobflow/core/flow.py
index 13193bdc..82a61691 100644
--- a/src/jobflow/core/flow.py
+++ b/src/jobflow/core/flow.py
@@ -284,11 +284,8 @@ def output(self, output: Any):
             The output of the flow. These should come from the output of one
             or more of the jobs.
         """
-        output_is_normalized = getattr(self, "_output_is_normalized", False)
-        self._output_is_normalized = False
-
         if output is not None:
-            if not output_is_normalized and contains_flow_or_job(output):
+            if contains_flow_or_job(output):
                 warnings.warn(
                     f"Flow '{self.name}' contains a Flow or Job as an output. "
                     f"Usually the Flow output should be the output of a Job or "
@@ -992,9 +989,6 @@ def __init__(self, fn, *args, **kwargs):
                 stacklevel=2,
             )
         output = _normalize_job_or_flow(output)
-        # Avoid checking the entire output again in Flow.output; normalization has
-        # already replaced every nested Job and Flow.
-        self._output_is_normalized = True
 
         super().__init__(name=name, jobs=children_list, output=output)
 
diff --git a/tests/core/test_flow_decorator.py b/tests/core/test_flow_decorator.py
index 14747fd5..7dc4dc88 100644
--- a/tests/core/test_flow_decorator.py
+++ b/tests/core/test_flow_decorator.py
@@ -212,24 +212,6 @@ def my_flow():
     assert flow1.output == {"nested": (flow1.jobs[0].output,)}
 
 
-def test_flow_does_not_recheck_normalized_output(monkeypatch):
-    """Test normalized decorated-flow outputs are not scanned a second time."""
-    from jobflow import flow
-
-    def fail_if_called(_):
-        raise AssertionError("normalized output was checked again")
-
-    monkeypatch.setattr("jobflow.core.flow.contains_flow_or_job", fail_if_called)
-
-    @flow
-    def my_flow():
-        return {"nested": (add(1, 2),)}
-
-    flow1 = my_flow()
-
-    assert flow1.output == {"nested": (flow1.jobs[0].output,)}
-
-
 def test_flow_returns_list():
     """Test that a flow that returns a list of OutputReferences
     can be created and run."""

From 4aa94a5930b8e7b7cdb7588d377b3bbb184adee4 Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 13:07:30 -0400
Subject: [PATCH 12/15] perf: limit normalization to one container level

---
 src/jobflow/core/flow.py          | 13 +++++++----
 src/jobflow/core/job.py           |  7 ++++--
 tests/core/test_flow_decorator.py | 39 ++++++++++++++++++++-----------
 3 files changed, 39 insertions(+), 20 deletions(-)

diff --git a/src/jobflow/core/flow.py b/src/jobflow/core/flow.py
index 82a61691..17336de3 100644
--- a/src/jobflow/core/flow.py
+++ b/src/jobflow/core/flow.py
@@ -932,16 +932,21 @@ def get_flow(
 
 
 def _normalize_job_or_flow(value):
-    """Replace Jobs and Flows in nested containers with their outputs."""
+    """Replace Jobs and Flows at the top level of a value with their outputs."""
+    def normalize_item(item):
+        if isinstance(item, (jobflow.Job, jobflow.Flow)):
+            return item.output
+        return item
+
     if isinstance(value, (jobflow.Job, jobflow.Flow)):
         return value.output
     if isinstance(value, list):
-        return [_normalize_job_or_flow(item) for item in value]
+        return [normalize_item(item) for item in value]
     if isinstance(value, tuple):
-        return tuple(_normalize_job_or_flow(item) for item in value)
+        return tuple(normalize_item(item) for item in value)
     if isinstance(value, dict):
         return {
-            _normalize_job_or_flow(key): _normalize_job_or_flow(item)
+            normalize_item(key): normalize_item(item)
             for key, item in value.items()
         }
     return value
diff --git a/src/jobflow/core/job.py b/src/jobflow/core/job.py
index 94052125..e5496296 100644
--- a/src/jobflow/core/job.py
+++ b/src/jobflow/core/job.py
@@ -214,8 +214,11 @@ def get_job(*args, **kwargs) -> Job:
                         args = args[1:]
 
             if _current_flow_context.get() is not None:
-                args = _normalize_job_or_flow(args)
-                kwargs = _normalize_job_or_flow(kwargs)
+                args = tuple(_normalize_job_or_flow(arg) for arg in args)
+                kwargs = {
+                    key: _normalize_job_or_flow(value)
+                    for key, value in kwargs.items()
+                }
 
             return Job(
                 function=f, function_args=args, function_kwargs=kwargs, **job_kwargs
diff --git a/tests/core/test_flow_decorator.py b/tests/core/test_flow_decorator.py
index 7dc4dc88..30442f97 100644
--- a/tests/core/test_flow_decorator.py
+++ b/tests/core/test_flow_decorator.py
@@ -180,36 +180,47 @@ def workflow(a, b):
 
 
 def test_flow_resolves_job_inputs_to_outputs():
-    """Test that Jobs used as inputs inside a decorated flow resolve to outputs."""
+    """Test top-level container Jobs resolve without traversing nested containers."""
     from jobflow import flow, job
-    from jobflow.managers.local import run_locally
 
     @job
-    def combine(a, values):
-        return a + values["nested"][0]
+    def consume(values):
+        return values
 
     @flow
-    def my_flow(a, b):
-        sum_job = add(a, b)
-        return combine(a, {"nested": [sum_job]})
+    def my_flow():
+        source = add(1, 2)
+        list_job = consume([source])
+        tuple_job = consume((source,))
+        dict_job = consume({"source": source})
+        nested_job = consume([[source], [1, 2]])
+        return [
+            list_job.output,
+            tuple_job.output,
+            dict_job.output,
+            nested_job.output,
+        ]
 
-    flow1 = my_flow(1, 2)
-    result = run_locally(flow1, ensure_success=True)
+    flow1 = my_flow()
+    source = flow1.jobs[0]
 
-    assert result[flow1.output.uuid][1].output == 4
+    assert flow1.jobs[1].function_args == ([source.output],)
+    assert flow1.jobs[2].function_args == ((source.output,),)
+    assert flow1.jobs[3].function_args == ({"source": source.output},)
+    assert flow1.jobs[4].function_args == ([[source], [1, 2]],)
 
 
-def test_flow_normalizes_nested_job_outputs():
-    """Test Jobs nested in decorated flow outputs resolve to references."""
+def test_flow_normalizes_top_level_job_outputs():
+    """Test Jobs at the top level of a decorated flow output are normalized."""
     from jobflow import flow
 
     @flow
     def my_flow():
-        return {"nested": (add(1, 2),)}
+        return {"job": add(1, 2)}
 
     flow1 = my_flow()
 
-    assert flow1.output == {"nested": (flow1.jobs[0].output,)}
+    assert flow1.output == {"job": flow1.jobs[0].output}
 
 
 def test_flow_returns_list():

From fe94194fa39a24413831cbdbbd8ad07b2a402891 Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 13:11:45 -0400
Subject: [PATCH 13/15] test: focus container normalization coverage

---
 tests/core/test_flow_decorator.py | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/tests/core/test_flow_decorator.py b/tests/core/test_flow_decorator.py
index 30442f97..155f98c8 100644
--- a/tests/core/test_flow_decorator.py
+++ b/tests/core/test_flow_decorator.py
@@ -180,7 +180,7 @@ def workflow(a, b):
 
 
 def test_flow_resolves_job_inputs_to_outputs():
-    """Test top-level container Jobs resolve without traversing nested containers."""
+    """Test top-level container Jobs resolve to outputs."""
     from jobflow import flow, job
 
     @job
@@ -193,12 +193,10 @@ def my_flow():
         list_job = consume([source])
         tuple_job = consume((source,))
         dict_job = consume({"source": source})
-        nested_job = consume([[source], [1, 2]])
         return [
             list_job.output,
             tuple_job.output,
             dict_job.output,
-            nested_job.output,
         ]
 
     flow1 = my_flow()
@@ -207,7 +205,6 @@ def my_flow():
     assert flow1.jobs[1].function_args == ([source.output],)
     assert flow1.jobs[2].function_args == ((source.output,),)
     assert flow1.jobs[3].function_args == ({"source": source.output},)
-    assert flow1.jobs[4].function_args == ([[source], [1, 2]],)
 
 
 def test_flow_normalizes_top_level_job_outputs():

From 388eb111b8727d47ebb914eaa5b11e47025af875 Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Tue, 11 Aug 2026 13:18:20 -0400
Subject: [PATCH 14/15] style: apply ruff formatting

---
 src/jobflow/core/flow.py | 4 ++--
 src/jobflow/core/job.py  | 3 +--
 2 files changed, 3 insertions(+), 4 deletions(-)

diff --git a/src/jobflow/core/flow.py b/src/jobflow/core/flow.py
index 17336de3..edfd5375 100644
--- a/src/jobflow/core/flow.py
+++ b/src/jobflow/core/flow.py
@@ -933,6 +933,7 @@ def get_flow(
 
 def _normalize_job_or_flow(value):
     """Replace Jobs and Flows at the top level of a value with their outputs."""
+
     def normalize_item(item):
         if isinstance(item, (jobflow.Job, jobflow.Flow)):
             return item.output
@@ -946,8 +947,7 @@ def normalize_item(item):
         return tuple(normalize_item(item) for item in value)
     if isinstance(value, dict):
         return {
-            normalize_item(key): normalize_item(item)
-            for key, item in value.items()
+            normalize_item(key): normalize_item(item) for key, item in value.items()
         }
     return value
 
diff --git a/src/jobflow/core/job.py b/src/jobflow/core/job.py
index e5496296..34421ec9 100644
--- a/src/jobflow/core/job.py
+++ b/src/jobflow/core/job.py
@@ -216,8 +216,7 @@ def get_job(*args, **kwargs) -> Job:
             if _current_flow_context.get() is not None:
                 args = tuple(_normalize_job_or_flow(arg) for arg in args)
                 kwargs = {
-                    key: _normalize_job_or_flow(value)
-                    for key, value in kwargs.items()
+                    key: _normalize_job_or_flow(value) for key, value in kwargs.items()
                 }
 
             return Job(

From f03a33cf1166ccdf3b1f44708f071e836f8af4c5 Mon Sep 17 00:00:00 2001
From: Andrew Rosen <asrosen93@gmail.com>
Date: Thu, 20 Aug 2026 13:07:30 -0400
Subject: [PATCH 15/15] fix: address decorated flow review feedback

---
 src/jobflow/core/flow.py      | 30 +++++++-----------------------
 src/jobflow/core/job.py       | 24 ++++++------------------
 src/jobflow/utils/__init__.py |  1 +
 src/jobflow/utils/find.py     | 18 ++++++++++++++++++
 tests/core/test_job.py        | 21 ---------------------
 5 files changed, 32 insertions(+), 62 deletions(-)

diff --git a/src/jobflow/core/flow.py b/src/jobflow/core/flow.py
index edfd5375..c6c1e472 100644
--- a/src/jobflow/core/flow.py
+++ b/src/jobflow/core/flow.py
@@ -13,7 +13,12 @@
 
 import jobflow
 from jobflow.core.reference import find_and_get_references
-from jobflow.utils import ValueEnum, contains_flow_or_job, suid
+from jobflow.utils import (
+    ValueEnum,
+    contains_flow_or_job,
+    replace_job_or_flow_with_output,
+    suid,
+)
 
 if TYPE_CHECKING:
     from collections.abc import Iterator, Sequence
@@ -931,27 +936,6 @@ def get_flow(
     return flow
 
 
-def _normalize_job_or_flow(value):
-    """Replace Jobs and Flows at the top level of a value with their outputs."""
-
-    def normalize_item(item):
-        if isinstance(item, (jobflow.Job, jobflow.Flow)):
-            return item.output
-        return item
-
-    if isinstance(value, (jobflow.Job, jobflow.Flow)):
-        return value.output
-    if isinstance(value, list):
-        return [normalize_item(item) for item in value]
-    if isinstance(value, tuple):
-        return tuple(normalize_item(item) for item in value)
-    if isinstance(value, dict):
-        return {
-            normalize_item(key): normalize_item(item) for key, item in value.items()
-        }
-    return value
-
-
 class DecoratedFlow(Flow):
     """A DecoratedFlow is a Flow that is returned on using the @flow decorator."""
 
@@ -993,7 +977,7 @@ def __init__(self, fn, *args, **kwargs):
                 "check the outputs of your @flow decorated function.",
                 stacklevel=2,
             )
-        output = _normalize_job_or_flow(output)
+        output = replace_job_or_flow_with_output(output)
 
         super().__init__(name=name, jobs=children_list, output=output)
 
diff --git a/src/jobflow/core/job.py b/src/jobflow/core/job.py
index 34421ec9..47e8d6dd 100644
--- a/src/jobflow/core/job.py
+++ b/src/jobflow/core/job.py
@@ -11,8 +11,9 @@
 from monty.json import MSONable, jsanitize
 from typing_extensions import Self
 
-from jobflow.core.flow import _current_flow_context, _normalize_job_or_flow
+from jobflow.core.flow import _current_flow_context
 from jobflow.core.reference import OnMissing, OutputReference
+from jobflow.utils.find import replace_job_or_flow_with_output
 from jobflow.utils.uid import suid
 
 if typing.TYPE_CHECKING:
@@ -214,10 +215,8 @@ def get_job(*args, **kwargs) -> Job:
                         args = args[1:]
 
             if _current_flow_context.get() is not None:
-                args = tuple(_normalize_job_or_flow(arg) for arg in args)
-                kwargs = {
-                    key: _normalize_job_or_flow(value) for key, value in kwargs.items()
-                }
+                args = replace_job_or_flow_with_output(args)
+                kwargs = replace_job_or_flow_with_output(kwargs)
 
             return Job(
                 function=f, function_args=args, function_kwargs=kwargs, **job_kwargs
@@ -610,12 +609,7 @@ def run(self, store: jobflow.JobStore, job_dir: Path = None) -> Response:
         from datetime import datetime
 
         from jobflow import CURRENT_JOB
-        from jobflow.core.flow import (
-            Flow,
-            _normalize_job_or_flow,
-            flow_build_context,
-            get_flow,
-        )
+        from jobflow.core.flow import get_flow
         from jobflow.core.schemas import JobStoreDocument
 
         index_str = f", {self.index}" if self.index != 1 else ""
@@ -637,13 +631,7 @@ def run(self, store: jobflow.JobStore, job_dir: Path = None) -> Response:
         if bound is not None and not isinstance(bound, types.ModuleType):
             function = types.MethodType(function, bound)
 
-        children: list[Job | Flow] = []
-        with flow_build_context(children):
-            response = function(*self.function_args, **self.function_kwargs)
-
-        children = [child for child in children if child.host is None]
-        if children and not isinstance(response, Response):
-            response = Flow(jobs=children, output=_normalize_job_or_flow(response))
+        response = function(*self.function_args, **self.function_kwargs)
         response = Response.from_job_returns(
             response, self.output_schema, job_dir=job_dir
         )
diff --git a/src/jobflow/utils/__init__.py b/src/jobflow/utils/__init__.py
index ed6e1648..906c8121 100644
--- a/src/jobflow/utils/__init__.py
+++ b/src/jobflow/utils/__init__.py
@@ -5,6 +5,7 @@
     contains_flow_or_job,
     find_key,
     find_key_value,
+    replace_job_or_flow_with_output,
     update_in_dictionary,
 )
 from jobflow.utils.log import initialize_logger
diff --git a/src/jobflow/utils/find.py b/src/jobflow/utils/find.py
index 5dfc98e9..1c5c0240 100644
--- a/src/jobflow/utils/find.py
+++ b/src/jobflow/utils/find.py
@@ -11,6 +11,24 @@
     from monty.json import MSONable
 
 
+def replace_job_or_flow_with_output(value):
+    """Replace Jobs and Flows in a value with their outputs."""
+    from jobflow import Flow, Job
+
+    if isinstance(value, (Job, Flow)):
+        return value.output
+    if isinstance(value, list):
+        return [replace_job_or_flow_with_output(item) for item in value]
+    if isinstance(value, tuple):
+        return tuple(replace_job_or_flow_with_output(item) for item in value)
+    if isinstance(value, dict):
+        return {
+            replace_job_or_flow_with_output(key): replace_job_or_flow_with_output(item)
+            for key, item in value.items()
+        }
+    return value
+
+
 def find_key(
     d: dict[Hashable, Any] | list[Any],
     key: Hashable | type[MSONable],
diff --git a/tests/core/test_job.py b/tests/core/test_job.py
index d1c41e0d..9b43a61d 100644
--- a/tests/core/test_job.py
+++ b/tests/core/test_job.py
@@ -1446,24 +1446,3 @@ def add_configured(a, b):
         f"Expected job2.config.manager_config to be {{'key': 'original'}}, "
         f"but got {job2.config.manager_config!r} — shared instance bug confirmed."
     )
-
-
-def test_job_collects_jobs_created_during_run(memory_jobstore):
-    """Test a job retains all jobs created while its function is running."""
-    from jobflow import job
-
-    @job
-    def add_job(a, b):
-        return a + b
-
-    @job
-    def create_jobs():
-        first = add_job(1, 2)
-        return add_job(first.output, 3)
-
-    response = create_jobs().run(memory_jobstore)
-
-    # The replacement contains both created jobs and the output-mapping job added by
-    # prepare_replace. Its output should refer to the second created job.
-    assert len(response.replace) == 3
-    assert response.replace.output.uuid == response.replace[-2].uuid