.. _managing-complex-workflows: Managing Complex Workflows ============================================== As a project grows, the expensive part is rarely the final result — it's the *granular* work underneath it. Fetching data from a slow or paid API. Training a model once per hyperparameter setting. Re-running that work every time you tweak something downstream is what makes iteration painful. oryxflow's answer is to make each granular unit its own **task**, cached independently. The engine tracks completeness per task, so when you change one thing it re-runs exactly what changed — and leaves the expensive, unchanged work alone. This page walks through the pattern data scientists reach for constantly: #. **Iterate granular** — one task per item (per hyperparameter, per state, per input file), each cached on its own. #. **Aggregate** — combine those outputs one level up into a single result. #. **Selectively reset** — when something the engine *can't* see changes (a data file, an external API), invalidate just that family of tasks and let everything downstream recompute. The expensive leaves stay cached. (Parameter and code changes need no reset — they rerun on their own; see below.) Worked example: hyperparameter tuning ------------------------------------------------------------ The classic case: load a dataset once (expensive), train a model once per hyperparameter value (the granular loop), then aggregate the scores to pick a winner. .. code-block:: python import oryxflow import pandas as pd oryxflow.set_dir('data/') ALPHAS = [0.01, 0.1, 1.0, 10.0] # the grid you iterate over — your own domain data class LoadData(oryxflow.tasks.TaskCachePandas): """Expensive: a slow/paid data pull you never want to repeat needlessly.""" def run(self): self.save(load_training_data()) # imagine a long API call @oryxflow.requires(LoadData) # wires requires(), copies params class TrainModel(oryxflow.tasks.TaskCachePandas): """One granular task per hyperparameter value. Cached independently.""" alpha = oryxflow.FloatParameter() def run(self): df = self.inputLoad() # the shared dataset, loaded once score = fit_and_score(df, alpha=self.alpha) self.save(pd.DataFrame({'alpha': [self.alpha], 'score': [score]})) class Tune(oryxflow.tasks.TaskCachePandas): """Aggregate: stack every trained model's score into one frame.""" def requires(self): return {a: TrainModel(alpha=a) for a in ALPHAS} def run(self): df = self.inputLoadConcat() # one row per alpha; 'alpha' column tags each self.save(df.sort_values('score', ascending=False)) Run it and read off the best hyperparameter: .. code-block:: python flow = oryxflow.Workflow(Tune) flow.run() results = flow.outputLoad() # every alpha's score, one frame best_alpha = results.iloc[0]['alpha'] The ``requires()`` dict is what fans the DAG out into one ``TrainModel`` per ``alpha``, and ``self.inputLoadConcat()`` stacks their outputs, tagging each row with that task's parameters so your ``alpha`` column survives the concat. (The mechanics of dict-``requires()`` and ``inputLoadConcat`` are covered as reference in :doc:`Advanced: Dynamic Tasks `.) Extend the grid — nothing wasted ------------------------------------------------------------ Decide you want to try more values? Add them to ``ALPHAS`` and run again: .. code-block:: python ALPHAS = [0.01, 0.1, 1.0, 10.0, 100.0] # added 100.0 flow.run() Only ``TrainModel(alpha=100.0)`` runs — the four you already trained are complete and are skipped, and the expensive ``LoadData`` is not touched at all. ``Tune`` recomputes because it now has a new input. You didn't reset anything; adding a parameter value is enough. What reruns, and when you do nothing ------------------------------------------------------------ A task is "complete" when its output exists **for its current parameters and its current code**. Two of the three things that can make a result stale are handled for you, no action required: * **You changed a parameter.** The task id is derived from the parameters, so a new value is a new id — no output yet, the task runs, and the old output for the old parameters is left untouched beside it. Add a value to the grid (a new ``alpha``) and *only* that new task runs. * **You changed the code.** oryxflow fingerprints each task's logic (next section). Edit a task's ``run()`` — or a helper it calls — and the task plus everything downstream recomputes on the next run. The third — a change the engine *can't* see, like a new data file or an API response — is the only one you drive by hand, with ``reset()`` (see :ref:`reset-no-code-token` below). Because completeness is checked recursively down the graph, an aggregator recomputes automatically whenever any of its inputs changes or a new one appears, with no bookkeeping on your part. .. _code-versioning: Automatic code invalidation ------------------------------------------------------------ Suppose you improve ``TrainModel`` — add a second scoring metric to its ``run()``. The parameters didn't change, so a parameters-only cache would happily serve the stale output. oryxflow instead **fingerprints the code**: on the next run every ``TrainModel`` recomputes, and so does everything downstream of it (``Tune``), overwriting in place at the same paths. ``LoadData`` is untouched. You did nothing — no reset, no version to bump. The fingerprint covers each task's own class **and every project-local symbol it references**, followed transitively across your modules — so editing a *helper function* the task calls is caught too, where the change actually lives, not just in ``run()``. Two properties make this safe to leave on: * **Cosmetic edits never recompute.** Files are hashed after AST normalization, so comments, docstrings, and formatting produce no fingerprint change at all. * **Granularity is per symbol, not per file.** One monolithic ``tasks.py`` is fine: editing one task's ``run()`` reruns that task alone, and editing a shared helper recomputes exactly the tasks that reference it (directly or through other helpers). The rerun reason even names the changed symbol — ``code change (auto: tasks.py::TrainModel)``. Referencing another *task* in ``requires()`` is dependency wiring, not a code reference — it never pulls that task's body into your fingerprint; the dependency cascade carries that staleness instead. What symbol analysis can't split apart (module-level side effects, star imports, dynamically created classes) falls back to whole-file granularity — extra reruns at worst, never a missed one. **Expensive tasks warn instead of silently recomputing.** A rerun overwrites the old output, so burning a long computation should be a decision, not a side effect of a refactor. A task whose last run took longer than ``settings.code_version_auto_expensive_s`` (default 600 seconds) is held complete when its code changes, and the run **warns** instead — answer it with one of the :ref:`three exits ` (reset, ``accept_code``, or pin it with ``code_version``). Set the threshold to ``None`` to make every code change recompute. To turn automatic tracking off entirely — parameters-only identity plus explicit pins — set ``oryxflow.settings.code_version_auto = False``. .. _code-blind-spots: When you changed something but nothing reran ------------------------------------------------------------ The fingerprint follows Python ``import`` statements under your project root. It **cannot** see: data-file contents, external APIs, installed packages, dynamic imports or monkeypatching, and tasks defined in a notebook or REPL (which aren't hashable at all). A change in any of those is invisible to the hash, so the task keeps its cache. The symptom is a **skip you didn't expect**: you changed something, ran, and the summary says ``0 ran``. Make this one check a habit — after any change you expect to recompute, confirm the affected tasks appear in ``result.ran`` (or ``oryxflow.events.runs()``) with a matching reason such as ``code change (auto: pipeline/train.py::TrainModel)``: * ``ran=0`` **after a change** means the change is in a blind spot. ``reset()`` the task that ingests it, or give that task an explicit ``code_version`` if it happens repeatedly. * ``ran=0`` **on an untouched pipeline** is the healthy "cache is trusted" signal. A blind spot never produces a *false* rerun or a false "verified unchanged" — it just degrades to parameters-only identity, the same trust level every cache file has always had. Pinning a task: the explicit ``code_version`` ------------------------------------------------------------ Sometimes automatic is the wrong sensitivity — a training task so expensive that a refactor-triggered recompute must be a deliberate decision, or logic the hash can't see (behavior driven by a config file, dynamic dispatch). Declare an explicit ``code_version`` and that task's own logic is **pinned**: it recomputes only when you bump the token, and a code edit *without* a bump produces a staleness warning instead of a rerun. .. code-block:: python @oryxflow.requires(LoadData) class TrainModel(oryxflow.tasks.TaskCachePandas): code_version = 2 # pinned: bump deliberately to recompute alpha = oryxflow.FloatParameter() def run(self): ... The token can be an int or a string (``'v2-log-features'``). Pins are **free to toggle**: the ``code_version`` line is itself stripped by AST normalization (like a comment), so adding, removing, or bumping it is never itself a source change. Under the hood oryxflow stores both the token and the code hash at each run and compares whichever matches the current mode — which is what gives the toggle these useful properties: * Pinning a task whose code is unchanged costs nothing — no recompute. * Pinning *in the same edit as a logic change* still recomputes: the hash catches what the pin would otherwise have blessed. * Unpinning just resumes automatic tracking; if you edited while pinned and never bumped, that one masked edit is caught and recomputes once. * Toggling never ripples downstream — dependents key on whether an upstream actually *rematerialized*, not on which mode tracked it. Pinned and automatic tasks mix freely in one pipeline, and a pinned task still reruns when its *upstream* recomputes — the pin covers its own logic, not its inputs. Why pin at all, if automatic needs nothing? The token makes the cache decision — "this logic changed, downstream must recompute" — an explicit, diffable part of the same edit, visible in code review and ``git log``. That's why AI-agent projects often prefer pins on their key tasks. .. note:: **One first-time trap.** Bringing an output under tracking for the *first* time — a pre-upgrade artifact, or the first pin in a ``code_version_auto = False`` project — in the *same edit* that changes its logic can bless the stale output, because there is no prior record to compare against. oryxflow warns when it can detect this (``output predates current code``); answer it with ``reset()`` to recompute, or ``flow.accept_code()`` if the outputs really are current. Once a task has one record on disk, the trap is closed for good — the stored hashes expose any edit no matter how you toggle the pin. .. _code-three-exits: The three exits for any code change ------------------------------------------------------------ Every code change — whether it auto-recomputed or a pin warned — has the same three exits, and they are **not** equal in risk: * **Recompute** — for automatic tasks, just run (it already happens); for pinned tasks, bump ``code_version``. Safe by default. * **Reset** — ``flow.reset(...)`` recomputes regardless, with no version bookkeeping. Also safe. * **Accept** — the change is output-equivalent (a rename, an extracted function, an added log line): ``flow.accept_code()`` (or ``oryxflow.accept_code(task)``) re-stamps the current code as accepted *without* rerunning. For an automatic task this is how you *skip* a recompute it would otherwise do. **Accept is the one exit that can silently bless a stale output** — use it only when you're certain; when unsure, recompute (cheap insurance; a wrongly-blessed cache is not). It accepts the task **and its whole upstream tree**, so call it on the most-downstream task you judge equivalent, and it prints a one-line summary of what it re-stamped — "nothing accepted" means it didn't reach the target. A bare ``flow.accept_code()`` covers your whole pipeline — every task it can compute, whichever final it hangs from — so one call blesses a multi-final pipeline, and it works from a fresh script (a one-shot bless after an upgrade needs no prior run). You can also name the tasks yourself, one or several: ``flow.accept_code([FinalA, FinalB])``. Prefer the instance/``flow`` form over the bare/class form: it is the only one that can stamp a baseline for record-less outputs, and anything the other forms miss simply recomputes (the safe direction). On a ``WorkflowMulti`` use ``flow.accept_code()`` (all flows, or one with ``flow=...``) — the module-level form doesn't know your flows' parameters. Tasks a flow reaches only *dynamically* (yielded inside a ``run()``) aren't in the static tree; accept those with an explicit instance if they warn. .. note:: The staleness warning for a pinned task prints without ``enable_logging()``:: StalenessWarning: task TrainModel: pipeline/train.py::TrainModel changed since cached run; code_version still 2 -- reusing cached output. Bump code_version to recompute, or oryxflow.accept_code(TrainModel) only if certain the output is equivalent. It dedupes per process — one line per distinct message, however many flows or parameter values hit it — and re-arms when the condition changes or the tasks rerun. ``result.warnings`` lists each distinct warning once, so its length answers "how many pending warnings do I have?" no matter how many flows or parameter values are involved. ``oryxflow.events.warnings()`` still records every occurrence, so even if your environment suppresses the print the record is never lost, and a strict ``-W error`` setup won't turn the advisory into a failed build. Which verb, when ------------------------------------------------------------ .. list-table:: :header-rows: 1 :widths: 42 24 34 * - I changed… - Do this - Why * - a value/knob that is a ``Parameter`` - nothing - new identity auto-reruns; old output kept side by side * - logic (a task's ``run()`` or a helper it imports), output will differ - nothing — just run - the code fingerprint moved; the task and everything downstream recompute * - logic of a **pinned** task (one that declares ``code_version``) - bump ``code_version`` - the pin is the authority; without a bump you get a warning, not a rerun * - logic of an **expensive** task (last run > ``code_version_auto_expensive_s``) - answer the warning: ``reset()`` / ``accept_code`` / pin - the guard holds it cached so a refactor can't silently burn a long run * - code, but output is provably identical (rename, extract, log line) - ``flow.accept_code()`` — only if certain; when unsure, recompute - re-stamps without recompute; the one non-recomputing exit * - nothing the hash can see, but I need a fresh compute (data file changed, suspect cache) - ``reset()`` - invisible to the fingerprint; force it at the task that ingests the change * - something, but the run skipped it (``0 ran`` after an edit) - ``reset()`` — or pin that task with ``code_version`` - the change is in a hash blind spot; treat the skip as a signal, not a convenience * - I want the outputs gone - ``reset()`` - delete * - pin added or removed (code untouched) - nothing - the record carries both dimensions; a pure mode flip never recomputes or ripples * - first time bringing an **untracked** output under tracking right after editing it - ``reset()`` once - no record exists yet, so grandfathering would bless the stale output (once a record exists this trap is caught automatically) * - nothing — but pre-existing outputs warn ``output predates current code`` - ``flow.accept_code()`` if the outputs are current, else ``reset()`` - the guard can't tell a fresh checkout from a stale cache; accepting stamps record-less outputs a baseline in one call Keeping old versions side by side ------------------------------------------------------------ By default a bump overwrites in place. To keep previous versions at readable paths — the compare-two-versions workflow — set ``keep_versions = True`` on the task (ideally with a string version): .. code-block:: python class TrainModel(oryxflow.tasks.TaskCachePandas): code_version = 'v2-log-features' keep_versions = True # outputs live at data/TrainModel/v2-log-features/... Bumping then writes to a new ``v.../`` directory and the old one stays on disk. ``keep_versions`` keys off the explicit token only — automatically-tracked tasks (no ``code_version``) always overwrite in place. Note that *turning on* ``keep_versions`` for an existing task relocates its output path (it gains the version segment), so the task recomputes once and the old non-versioned artifact is left behind — it's not a transparent toggle. .. _reset-no-code-token: Selectively reset when code versioning doesn't apply ------------------------------------------------------------ ``reset`` remains the right tool when there's no code token to move: a data file changed, you suspect a corrupt cache, or you simply want outputs deleted. The point is to reset **only** the family that changed, so the shared ``LoadData`` stays cached: .. code-block:: python flow = oryxflow.Workflow(Tune) flow.reset_upstream(Tune, only=TrainModel) # every TrainModel in the DAG, nothing else flow.run() ``reset_upstream(Tune, only=TrainModel)`` walks the whole graph upstream of ``Tune`` and invalidates just the ``TrainModel`` instances — every ``alpha`` at once, found via the DAG so you never hand-list them. ``LoadData`` is a *different* family and is left complete, so the slow data pull does **not** repeat. On the next ``run()`` all four models retrain and ``Tune`` recomputes on top (recursive completeness), while ``LoadData`` is served from cache. Contrast the three reset scopes you'll actually use: * ``flow.reset(TrainModel(alpha=0.1))`` — one specific instance. Use when you're debugging a single case. * ``flow.reset_upstream(Tune, only=TrainModel)`` — one *family*, everywhere it appears upstream. Use when something the code hash can't see changed for that family and you want every instance recomputed, cheap dependencies preserved. * ``flow.reset_upstream(Tune)`` — the whole upstream, including ``LoadData``. Use only when the raw inputs themselves are stale. This is the core discipline: **reset at the level you changed, not above it.** It's what keeps a tweak to a fast step from triggering hours of expensive re-fetching or re-computation. The event stream: what ran, when, and why ------------------------------------------------------------ Every run appends structured events to a plain-text JSONL stream — always on, written asynchronously (microseconds), so the record exists even when your script discards the run result. The file layout is a contract you can script against: * current month: ``.oryxflow/events.jsonl`` (a stable head — ``tail -30`` always shows the latest) * offloaded months: ``.oryxflow/events-YYYYMM.jsonl`` (immutable once offloaded) * all history: glob ``.oryxflow/events*.jsonl`` Events include ``run_started`` / ``task_ran`` / ``task_failed`` / ``run_finished`` / ``code_warning`` / ``code_accepted`` / ``task_log``. Each ``task_ran`` carries the full recipe — params, ``code_version``, code fingerprint, source hashes, git SHA, duration — and the **reason** it ran (``output missing`` / ``code change (auto: pipeline/train.py)`` / ``code change (1 -> 2)`` / ``upstream rerun``), so "why did this recompute?" and "was this produced by current code?" are queries, not guesses. Anything a task logs via ``self.logger`` is captured as a ``task_log`` event — log your decision-relevant scalars (``self.logger.info("corr_avg={}", corr)``) and they become next session's memory. .. code-block:: python oryxflow.events.print_status() # session-start orientation, printed: pending code # warnings, last run per family, recent failures oryxflow.events.status() # the same as data (a dict) -- it returns, doesn't print oryxflow.events.runs(task_family='TrainModel', last=2) # diff params/code_version/hashes oryxflow.events.runs(flow='Retail') # per-flow when using WorkflowMulti From the shell, no Python needed:: tail -30 .oryxflow/events.jsonl grep TrainModel .oryxflow/events*.jsonl jq 'select(.type=="task_ran") | {task_id, reason, duration_s}' .oryxflow/events.jsonl ``run()`` returns the same story in memory: ``result.reasons`` maps each task that ran to why, and ``result.warnings`` lists unacknowledged code changes. Add ``.oryxflow/`` to your ``.gitignore`` — run records are high-frequency exhaust, not source. Disable entirely with ``oryxflow.settings.events = False``. Where the freshness records live ------------------------------------------------------------ The code fingerprints live in one small JSON file per data directory (``/.oryxflow-code-status.json``). It describes those exact artifacts, so **move or restore a data directory whole** — file and artifacts together, always via the same channel. An artifact without a record is simply treated as current (the same trust level every file has today); partial restores are on you. Correctness depends only on this file, never on the event log. What caching does *not* protect against ------------------------------------------------------------ Everything above solves one family of problems — the *mechanical* one: staleness, provenance, and memory. That family genuinely yields to tooling, and a cached result you've kept fresh is worth trusting *as a computation*. But be clear about the boundary: oryxflow guarantees a result was produced by the code and inputs it records — **not** that the result is correct. A perfectly versioned, fully reproducible pipeline can still hand you a wrong number, and nothing here will flag it: * a join that should be many-to-one runs many-to-many and inflates every downstream aggregate; * a percentage computed against the wrong denominator, or a ``groupby`` that silently drops null-keyed rows; * a dtype coercion that eats a ZIP code's leading zero, or a timezone shift that moves rows a day; * a backtest that peeks at the future, or a correlation read off overlapping windows as if the points were independent; * a number quoted from memory or eyeballed off a chart instead of from the saved artifact. None of these raise; each completes cleanly and prints something plausible. Versioning the code just makes the wrong join *reproducible*. These are caught by **habit, not machinery** — validate the merge and assert the row relationship, look at the frame's shape and null counts before stating a finding, quote every number from an artifact you can re-open. (Note this is a different thing from the hash blind spots above: a changed data file is invisible but has a verb — ``reset`` the loader; a wrong join has no verb, only vigilance.) That is why the :doc:`Claude Code plugin ` ships those conventions alongside the library, delivered at the point you're writing the task rather than in a review afterward. And the judgment calls — is this method right for the question, is this effect within noise, does the data actually behave the way I assumed — are yours; no cache decides them. .. _claude-md-snippet: CLAUDE.md snippet for AI-agent projects ------------------------------------------------------------ For projects driven by AI coding agents, the recommended setup is the :doc:`oryxflow Claude Code plugin ` — it ships these rules (and more: project scaffolding, task templates, conventions) as a skill that loads automatically, stays current with the library, and needs no per-project copy. If you can't use the plugin (a different agent, a locked-down environment), paste this snapshot of the rules into the project's ``CLAUDE.md``: .. code-block:: markdown ## oryxflow cache & provenance rules 1. Session start / after `/clear`: call `oryxflow.events.print_status()` — pending code warnings, last run per task family, recent failures — before assuming anything about cache state (`events.status()` returns the same as a dict for filtering; it prints nothing). No-Python fallback: `tail -30 .oryxflow/events.jsonl`. 2. Editing task or helper logic needs NO cache action — code invalidation is automatic (AST-hash of the task's own class + the repo-local symbols it references, transitively; comments/formatting never count, and editing an unrelated task in the same file never reruns its siblings — one monolithic tasks.py is fine) and propagates downstream. Do not hand-chain `reset()` calls for code changes. Expensive tasks (last run > `settings.code_version_auto_expensive_s`, default 600s) warn instead of silently recomputing — answer with reset / accept_code / pin. Exception: a task that declares `code_version` is PINNED — automatic tracking of its own logic is off, so bump its token **in the same edit**. Pins toggle freely: adding or removing one on unchanged code never recomputes (and never ripples downstream); an edit masked while pinned is caught the moment the pin comes off. 3. **Verify the rerun happened.** After any code edit, the next run must show the edited band in `result.ran` (or `oryxflow.events.runs()`) with a matching reason — `code change (auto: ::)` or `code change (1 -> 2)`. `ran=0` after an edit means the hash didn't see the change (data file, installed package, dynamic call, notebook-defined task): `reset()` the affected task, or pin it with `code_version` if it recurs. `ran=0` on an untouched pipeline is the healthy signal. 4. Expensive recompute you judge output-equivalent (pure refactor): `flow.accept_code()` / `oryxflow.accept_code(anchor_task)` re-stamps the task and its whole upstream tree without rerunning — only if certain; when unsure, let it rerun. `preview()` first to see the pending band. A bare `flow.accept_code()` covers the whole pipeline — every task the flow can compute, multi-final pipelines included, from a fresh process; a list also works (`flow.accept_code([FinalA, FinalB])`). It prints what it re-stamped — "nothing accepted" means it didn't reach the target (use the instance/flow form, not the class/bare form). An `output predates current code` warning (outputs with no record yet, e.g. after an upgrade) is answered the same way: `flow.accept_code()` if the outputs are current, `reset()` if not. On WorkflowMulti use `flow.accept_code()` (all flows) — the module-level bulk form doesn't know the flows' parameters. Answer every staleness warning with one of its exits — bump, accept, or reset. Never leave one firing across runs. 5. After a run, read the returned result: `result.reasons` says why each task ran; `result.warnings` lists unacknowledged code changes (each distinct warning once, so its length is the pending count). Never hand-roll aggregation — `MultiRunResult` exposes `.ran`/`.complete`/`.failed`/`.reasons`/`.warnings` across flows, and the per-build verdict is logged durably as `run_finished` events. 6. "The numbers changed and I don't know why": compare the last two runs — `oryxflow.events.runs(task_family='TaskX', last=2)` — and diff params, code_version, source_hashes. 7. Log decision-relevant scalars inside `run()` via `self.logger.info(...)` — they're captured as `task_log` events and become next session's memory. 8. Experiments you want side by side: string `code_version` plus `keep_versions = True` (explicit token only — auto-tracked tasks overwrite in place). 9. Raw stream: current = `.oryxflow/events.jsonl`; offloaded months = `events-YYYYMM.jsonl`; all history = glob `events*.jsonl`. Prefer `events.runs()`/`status()` in Python. 10. Data-file or external-API changes are invisible to the code hash — `reset()` the task that ingests them (a downstream reset re-loads the cached old input). Scaling up: hierarchies and independent experiments ------------------------------------------------------------ The same three moves scale in two directions. **Deeper hierarchies.** Aggregators compose: a per-state task feeds a per-country task feeds a per-sector task, each level a ``requires()`` fan-out combined with ``inputLoadConcat()``. The whole tree is still one DAG, so preview, the run summary, and selective reset all reach every level. **Independent experiments.** When you want to manage several runs separately — each with its own output and its own reset scope — drive the top with :doc:`WorkflowMulti ` over a params grid instead of one more aggregator, then combine across flows with ``outputLoadConcat``: .. code-block:: python flow = oryxflow.WorkflowMulti(Sector, params={'sector': ['Retail', 'Office']}) flow.run() dfall = flow.outputLoadConcat(Sector) # all sectors, one tagged frame flow.reset_upstream(Sector, only=CountryFeatures) # reset one family across every flow A complete, runnable version of this multi-level dev loop — iterate on one ``(sector, country)`` first, then roll the change out to every flow *without re-fetching the expensive per-state source* — is in ``docs/example-flow-multi.py``. The full reference for dict-``requires()``, ``inputLoadConcat``, ``outputLoadConcat``, and the ``only=`` reset filter is :doc:`Advanced: Dynamic Tasks `. .. tip:: This is exactly what the :doc:`Claude Code plugin ` is built to manage. Describe the hierarchy in plain language and it writes the fan-out ``requires()`` and the ``inputLoadConcat()`` aggregators; when you iterate, it scopes the reset for you — resetting just the family you changed (``reset_upstream(..., only=...)``) so the expensive leaf tasks are preserved.