Skip to content

feat: simplify apply_tool_multiplex() using multiplex --flat mode #869

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
k-rister merged 1 commit into from
Aug 17, 2026
Merged

feat: simplify apply_tool_multiplex() using multiplex --flat mode #869

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 17 additions & 55 deletions rickshaw-run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1324,58 +1324,37 @@ def load_tool_params(self):

def apply_tool_multiplex(self, tool_entry, tool_dir):
"""If this tool declares a multiplex.json, run its flat tool-params.json
params through multiplex.py (unmodified) for validation/transform/preset
params through multiplex.py's --flat mode for validation/transform/preset
support identical to benchmarks. No-op if the tool has no multiplex.json
(fully backward compatible for tools that haven't opted in).

Tools have no sets/include/cartesian-product concept -- they always
have exactly one implicit parameter set and (per schema/tool-params.json)
never more than one value per param -- so this uses multiplex's --flat
mode rather than wrapping into the general sets/global-options document
benchmarks use. Filtering of disabled params and enforcing "exactly one
combination" both happen inside multiplex.py itself now; any violation
surfaces as a non-zero rc below, same as any other multiplex failure.
"""
requirements_file = os.path.join(tool_dir, "multiplex.json")
if not os.path.exists(requirements_file):
return

tool_id = tool_entry["tool-id"]

# WRAP: tool-params.json's flat {"arg","val"} shape -> multiplex's
# "sets" shape. Every value becomes a one-element "vals" array, which
# makes itertools.product() (multiplex's actual multiplication step)
# structurally incapable of producing more than one combination --
# this is what makes reusing multiplex.py unmodified safe for tools.
#
# Pre-filter disabled params ourselves (rather than relying on
# multiplex's own enabled-filtering) since multiplex's essentials-
# preset-override runs BEFORE its enabled-filtering step and could
# otherwise silently resurrect a param the user explicitly disabled.
wrapped_params = [
{
"arg": p["arg"],
"vals": [p["val"]],
"role": "all", # tools have no client/server concept; set
# explicitly rather than relying on multiplex's
# sanitize_set() default of "client", which
# could cause spurious dedup mismatches
# against presets that specify "role": "server"
}
for p in tool_entry.get("params", [])
if p.get("enabled") != "no"
]

mv_params_doc = {
"schema": {"version": "2021.12.02"},
"global-options": [],
"sets": [{"params": wrapped_params}],
}
mv_params_file = os.path.join(self.config_dir, f"{tool_id}-mv-params.json")
flat_params_file = os.path.join(self.config_dir, f"{tool_id}-flat-params.json")
# toolbox.json.save_json_file() unconditionally xz-compresses and
# renames to "<file>.xz" with no way to opt out -- multiplex.py's own
# file reading has no decompression support, so a plain write is
# required here (matching how the benchmark mv-params path writes its
# own equivalent file directly rather than via save_json_file()).
with open(mv_params_file, "w") as f:
json.dump(mv_params_doc, f)
with open(flat_params_file, "w") as f:
json.dump(tool_entry.get("params", []), f)

tool_params_out_file = os.path.join(self.config_dir, f"{tool_id}-tool-params-multiplexed.json")
multiplex_cmd = (
f"{os.environ.get('MULTIPLEX_HOME', '')}/multiplex.py "
f"--input {mv_params_file} --output {tool_params_out_file} "
f"{os.environ.get('MULTIPLEX_HOME', '')}/multiplex.py --flat "
f"--input {flat_params_file} --output {tool_params_out_file} "
f"--requirements {requirements_file}"
)
logger.debug("about to run: %s", multiplex_cmd)
Expand All @@ -1385,29 +1364,12 @@ def apply_tool_multiplex(self, tool_entry, tool_dir):
logger.error("multiplex output is:\n%s", output)
sys.exit(1)

# UNWRAP: multiplex output -> tool-params.json's flat shape. Only
# arg/val survive -- role/id have no meaning for tools and would
# violate tool-params.json's additionalProperties:false if kept.
multiplexed_sets, err = load_json_file(tool_params_out_file)
if multiplexed_sets is None:
multiplexed_params, err = load_json_file(tool_params_out_file)
if multiplexed_params is None:
logger.error("[ERROR] could not load multiplexed tool params for '%s': %s", tool_id, err)
sys.exit(1)

if len(multiplexed_sets) != 1:
# Should be structurally impossible given the wrap above -- fail
# loudly instead of silently taking [0] if a tool's multiplex.json
# ever violates that assumption (e.g. via a misused "include").
logger.error(
"[ERROR] tool '%s' multiplex.json produced %d parameter combinations; "
"tool params must resolve to exactly one combination",
tool_id, len(multiplexed_sets),
)
sys.exit(1)

tool_entry["params"] = [
{"arg": param["arg"], "val": param["val"]}
for param in multiplexed_sets[0]
]
tool_entry["params"] = multiplexed_params

def load_utility_params(self):
for utility in UTILITIES:
Expand Down
68 changes: 32 additions & 36 deletions tests/test_apply_tool_multiplex.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,14 +4,16 @@

"""Unit tests for rickshaw-run.py's RunState.apply_tool_multiplex() -- the
bridge that lets a tool's flat tool-params.json params get validated/
transformed/preset-applied by multiplex.py, identical to benchmarks, without
ever risking real parameter multiplication (crucible#653).
transformed/preset-applied by multiplex.py's --flat mode, identical in
spirit to benchmarks but without ever needing sets/include/cartesian-
product (tools always have exactly one implicit param set and, per
schema/tool-params.json, never more than one value per param).

toolbox is mocked out rather than required, since rickshaw-run.py imports
from it at module scope and CI does not check toolbox out for this job.
multiplex.py itself is not invoked -- run_cmd is mocked so these tests
exercise only the wrap/unwrap logic in apply_tool_multiplex(), not
multiplex.py's own internals (which are that project's own responsibility).
exercise only apply_tool_multiplex()'s own logic, not multiplex.py's --flat
mode internals (which are that project's own responsibility).
"""

import importlib.machinery
Expand Down Expand Up @@ -106,17 +108,17 @@ def fake_save_json_file(filename, data, schema_file=None):
return mod


def fake_multiplex(sets):
"""Build a run_cmd side_effect simulating multiplex.py: reads the --input
file this code wrote, writes back `sets` (a list of param lists) to the
--output path, and returns rc=0. Pass rc= to simulate a failure instead.
def fake_multiplex(params):
"""Build a run_cmd side_effect simulating multiplex.py --flat: reads the
--input file this code wrote, writes back `params` (a flat list of
{arg, val} dicts) to the --output path, and returns rc=0.
"""

def _side_effect(cmd):
match = re.search(r"--output (\S+)", cmd)
output_file = match.group(1)
with open(output_file, "w") as f:
json.dump(sets, f)
json.dump(params, f)
return (cmd, "", 0)

return _side_effect
Expand All @@ -141,31 +143,34 @@ def test_no_multiplex_json_is_noop(self):
self.assertEqual(tool_entry["params"], [{"arg": "interval", "val": "10"}])
self.run_cmd.assert_not_called()

def test_wraps_and_unwraps_round_trip(self):
def test_flat_round_trip(self):
self._write_multiplex_json()
self.run_cmd.side_effect = fake_multiplex(
[[{"arg": "interval", "val": "10", "role": "all"}]]
[{"arg": "interval", "val": "10"}]
)
tool_entry = {"tool-id": "kernel", "params": [{"arg": "interval", "val": "10"}]}
self.state.apply_tool_multiplex(tool_entry, self.tool_dir)

# role must not leak into the unwrapped result
self.assertEqual(tool_entry["params"], [{"arg": "interval", "val": "10"}])

# verify the wrapped input actually sent to multiplex used a
# one-element vals array and role "all", not multiplex's "client" default
cmd = self.run_cmd.call_args[0][0]
self.assertIn("--flat", cmd)

# the input sent to multiplex is the tool's params, flat and
# unmodified -- no wrap into a sets/global-options document, no
# injected "vals"/"role" (--flat mode owns that internally now)
input_file = re.search(r"--input (\S+)", cmd).group(1)
with open(input_file) as f:
wrapped = json.load(f)
self.assertEqual(
wrapped["sets"][0]["params"],
[{"arg": "interval", "vals": ["10"], "role": "all"}],
)

def test_disabled_param_excluded_from_wrap(self):
flat_input = json.load(f)
self.assertEqual(flat_input, [{"arg": "interval", "val": "10"}])

def test_disabled_param_passed_through_unfiltered(self):
# Filtering a disabled param is now multiplex --flat mode's own
# responsibility (param_enabled() inside apply_flat_params()) --
# rickshaw no longer pre-filters client-side, matching how the
# benchmark path has always relied on multiplex for this.
self._write_multiplex_json()
self.run_cmd.side_effect = fake_multiplex([[]])
self.run_cmd.side_effect = fake_multiplex([])
tool_entry = {
"tool-id": "kernel",
"params": [{"arg": "interval", "val": "10", "enabled": "no"}],
Expand All @@ -175,8 +180,11 @@ def test_disabled_param_excluded_from_wrap(self):
cmd = self.run_cmd.call_args[0][0]
input_file = re.search(r"--input (\S+)", cmd).group(1)
with open(input_file) as f:
wrapped = json.load(f)
self.assertEqual(wrapped["sets"][0]["params"], [])
flat_input = json.load(f)
self.assertEqual(
flat_input,
[{"arg": "interval", "val": "10", "enabled": "no"}],
)

def test_multiplex_failure_exits(self):
self._write_multiplex_json()
Expand All @@ -199,18 +207,6 @@ def test_empty_set_fail_without_defaults_preset_exits(self):
with self.assertRaises(SystemExit):
self.state.apply_tool_multiplex(tool_entry, self.tool_dir)

def test_more_than_one_combination_is_rejected(self):
self._write_multiplex_json()
self.run_cmd.side_effect = fake_multiplex(
[
[{"arg": "interval", "val": "10", "role": "all"}],
[{"arg": "interval", "val": "20", "role": "all"}],
]
)
tool_entry = {"tool-id": "kernel", "params": [{"arg": "interval", "val": "10"}]}
with self.assertRaises(SystemExit):
self.state.apply_tool_multiplex(tool_entry, self.tool_dir)


if __name__ == "__main__":
unittest.main()
Loading