mirror of
https://github.com/run-llama/workflows-py.git
synced 2026-07-16 05:31:56 -04:00
Adrian Lyjak
8dc16b6987
`control_loop.py` was a single 2128-line module fusing two unrelated
concerns — the stateful async runtime and the pure reducer — with the
reducer dominated by two functions that read as walls:
`_process_step_result_tick` (~400 lines) and `_process_add_event_tick`
(~230 lines). You couldn't follow either in one pass.
This splits the module into a package along the reducer pattern's
natural seam and breaks the two god-functions into named phases. No
behavior change: the reducer is pure, so the existing transformation and
collection-stream suites pin command ordering and stream accounting.
## Package split
`control_loop.py` becomes `control_loop/`:
- `runner.py` — the async runtime: worker tasks, the scheduled-wakeup
heap, the adapter, turning reducer commands into side effects.
- `reduce.py` — the pure reducer: `_reduce_tick` and every per-tick
processor, plus retry/collect/replay helpers. `State + Tick -> (State,
Commands)`.
- `streams.py` — collection-stream accounting: fan-out stream lifecycle,
open-work-item counting, collect-batch release.
The dependency graph is a DAG: `runner -> reduce -> streams`. The one
back-edge (`streams`' release path needs `reduce`'s
`_add_or_enqueue_event`) is a single inline import at that chokepoint,
commented.
The package facade exports only the cross-package API — `control_loop`,
`rebuild_state_from_ticks`, `rebuild_state_from_ticks_stream`,
`replay_ticks_stream` — the symbols sibling packages (server, dbos,
`external_context`, `step_function`) actually import. White-box tests
import the internals they exercise straight from the owning submodule
(`.reduce` / `.runner` / `.streams`). `import time` stays in the facade
because tests patch `control_loop.time.time`, and the submodule loggers
keep the `workflows.runtime.control_loop` name so caplog filters still
match.
## Decomposing the reducer functions
Both now read top-to-bottom as a sequence of named steps.
`_process_step_result_tick`:
```python
this_execution = _find_in_progress(worker_state, tick.worker_id)
rerun = _rerun_for_stale_collect_buffer(tick, worker_state, this_execution)
if rerun is not None:
return state, rerun
scope = _fan_out_scope(state, step_name, this_execution, fanned_out=fanned_out)
for result in tick.result:
_apply_step_result(result, ..., acc=acc)
acc.commands.extend(_resolve_work_item_in_stream(tick, state, scope, acc, now_seconds))
# finalize: NOT_RUNNING transition, drop from in_progress, drain queue
```
The per-result flags that were locals threaded through 400 lines now
ride on a small `_StepResultAcc` carrier, and the ~120-line
retry/`catch_error` branch is extracted to
`_schedule_retry_or_route_failure`.
`_process_add_event_tick` splits into its three real phases:
`_redeliver_collection_payload` (payload-carrying ticks route straight
to their collect target and stop), `_resolve_waiters`, then
`_route_to_accepting_steps` — with the gnarly collect-member routing
pulled into `_route_member_to_collect_step`.
One rename: `_apply_stream_work_delta` -> `_adjust_open_work_items`,
matching the `open_work_items` field it mutates.