Plans do not function with `*args`

[EmsArnold]

Currently plans with *args do not function as expected.

A plan which broadly looks like:

def test_move(*args: Movable[T] | T, group: Group | None = None) -> MsgGenerator:
    return (yield from bps.mv(*args, group=group))

and is run from the client:

plans = bc.plans
plans.test_move(devices.sample_stage.x, "0")

you receive the following error:

2026-03-23 14:07:13,012     INFO blueapi.worker.task_worker Submitting: task_id='b3744ab9-b5a6-4846-ad06-060ba635a87c' task=Task(name='test_move', params={'args': 'sample_stage.x', 'group': '0'}, metadata={'user': 'vgz78127', 'instrument_session': 'cm44191-1', 'tiled_access_tags': ['{"proposal": 44191, "visit": 1, "beamline": "b01-1"}']}) request_id=None is_complete=False is_pending=True errors=[] outcome=None
2026-03-23 14:07:13,013     INFO blueapi.worker.task_worker Got new task: task_id='b3744ab9-b5a6-4846-ad06-060ba635a87c' task=Task(name='test_move', params={'args': 'sample_stage.x', 'group': '0'}, metadata={'user': 'vgz78127', 'instrument_session': 'cm44191-1', 'tiled_access_tags': ['{"proposal": 44191, "visit": 1, "beamline": "b01-1"}']}) request_id=None is_complete=False is_pending=True errors=[] outcome=None
2026-03-23 14:07:13,013     INFO blueapi.worker.task Asked to run plan test_move with {'args': 'sample_stage.x', 'group': '0'} and metadata {'user': 'vgz78127', 'instrument_session': 'cm44191-1', 'tiled_access_tags': ['{"proposal": 44191, "visit": 1, "beamline": "b01-1"}']} for all runs
2026-03-23 14:07:13,014    ERROR blueapi.worker.task_worker Task failed
2026-03-23 14:07:13,014    ERROR blueapi.worker.task_worker test_move() got an unexpected keyword argument 'args'
Traceback (most recent call last):
  File "/app/.venv/lib/python3.11/site-packages/blueapi/worker/task_worker.py", line 441, in process_task
    result = self._current.task.do_task(self._ctx)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/blueapi/worker/task.py", line 41, in do_task
    result = ctx.run_engine(func(**prepared_params))
                            ^^^^^^^^^^^^^^^^^^^^^^^
TypeError: test_move() got an unexpected keyword argument 'args'

If run from the client in a slightly different manner, e.g. plans.test_move(devices.sample_stage.x, 0) or plans.test_move(devices.sample_stage.x, 0, group = "foo"), you receive a blueapi.client.rest.InvalidParametersError with slightly differing flavours.

This causes some of the plans (such as move to have syntax which is significantly different from that of the underlying bluesky plan.

These wrapped plans and plan_stubs were originally written by @DiamondJoseph in #793, and the only mention of *args is that they shall not be used for plans and plan_stubs.

Acceptance Criteria

• Be able to use *args in plans, or have a reasoning/explanation for not using *args which is documented, perhaps in Write Bluesky Plans for Blueapi.

[oliwenmandiamond]

So there seems to be a deliberate choice to not support *args in plans according to those docs. I don't understand why this is the case because this means any generic scan command will now not be possible in the new system even though we already have scan plans that we could use.

[oliwenmandiamond]

So from your logs, we can see it is using your second value, the move position and putting it in the groups argument. I've tested to see if providing a list or args=[...] fixes it but same result
>>> plans.test_move([devs.fsj1, "Out"])
>>> plans.test_move(args=[devs.fsj1, "Out"])

2026-03-26 11:30:54,591     INFO blueapi.worker.task Asked to run plan test_move with {'args': ['fsj1', 'Out']} and metadata {'user': 'xol73553', 'instrument_session': 'cm12345-1'} for all runs
2026-03-26 11:30:54,592    ERROR blueapi.worker.task_worker Task failed
2026-03-26 11:30:54,592    ERROR blueapi.worker.task_worker test_move() got an unexpected keyword argument 'args'
Traceback (most recent call last):
  File "/app/.venv/lib/python3.11/site-packages/blueapi/worker/task_worker.py", line 441, in process_task
    result = self._current.task.do_task(self._ctx)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/blueapi/worker/task.py", line 41, in do_task
    result = ctx.run_engine(func(**prepared_params))
                            ^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/bluesky/utils/__init__.py", line 2015, in wrapper
    return Plan(bs_plan, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/bluesky/utils/__init__.py", line 1966, in __init__
    self._iter = f(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^
TypeError: test_move() got an unexpected keyword argument 'args'
Exception ignored in: <function Plan.__del__ at 0x7f5ef9fda480>
Traceback (most recent call last):
  File "/app/.venv/lib/python3.11/site-packages/bluesky/utils/__init__.py", line 1982, in __del__
    if self._stack:

We need to make it so if *args is used, blueapi must enforce anything after must be keyword argument and for it to even accepts args as valid.

[oliwenmandiamond]

This is a more fundamental framework issue as blueapi is only designed to work with kwargs rather than args and kwargs

blueapi/worker/task.py

    def do_task(self, ctx: BlueskyContext) -> None:
        ...
        func = ctx.plan_functions[self.name]
        prepared_params = self.prepare_params(ctx)
        ctx.run_engine.md.update(self.metadata)
        result = ctx.run_engine(func(**prepared_params))
        ...

the prepared_params runs the plan with every argument as kwargs so *args would never work. You would have to redesign task and TaskRequest to support

    args: list[Any] = []
    kwargs: dict[str, Any] = {}

rather than just parameters

    parameters: dict[str, Any] = {}

Then, we can support this by giving these arguments to the plan

        func = ctx.plan_functions[self.name]
        prepared_args, prepared_kwargs = self.prepare_params(ctx)
        ctx.run_engine.md.update(self.metadata)
        result = ctx.run_engine(func(*prepared_args, **prepared_kwargs))

In the meantime, the only alternative we can do is

@plan
def test_move(
    args: list[Movable[Any] | Any], group: str | None = None, wait: bool = True
) -> MsgGenerator:

which becomes

>>> plans.test_move([devs.fsj1.open, False])

rather than

@plan
def test_move(
    *args: Movable[Any], group: str | None = None, wait: bool = True
) -> MsgGenerator: