feat(cli): Add worker CLI with configurable options

[evanh]

Summary

Adds a taskbroker_client CLI for running workers, with a click option for every constructor argument of TaskWorker and PushTaskWorker.

• worker — runs a pull-mode TaskWorker, with an option for each TaskWorker.__init__ argument.
• push_worker — runs a push-mode PushTaskWorker, with an option for each PushTaskWorker.__init__ argument.
• spawn / scheduler — existing example helpers.

New tunables

• --grpc-max-message-size (on push_worker): threaded through a new grpc_max_message_size argument on PushTaskWorker. When unset, the worker falls back to the TASKBROKER_GRPC_MAX_MESSAGE_SIZE env var, so existing env-based deployment config keeps working.
• --track-producer-futures/--no-track-producer-futures (on both commands): arroyo only reads ARROYO_TRACK_PRODUCER_FUTURES from the environment, so the command sets that env var from the flag before starting the worker. It's on both commands because the future-tracking producer runs in child processes for both pull and push modes.

Test plan

• black, isort, flake8, and mypy pass via pre-commit.
• Verified both commands build and expose the new options, and that the arroyo flag round-trips through the env var arroyo reads (True → "True" enables, False → "False" disables).

🤖 Generated with Claude Code

[untitaker]

oh wait. can we have one cli for push/non-push? otherwise i think we might have a hard time unifying the templates too (each app will need to configure where the push cli is mounted, and where the non-push cli is mounted)

or a click group for push/non-push CLIs, and apps attack the click group to their main group

[evanh]

or a click group for push/non-push CLIs, and apps attack the click group to their main group

What do you mean by this? Isn't that what they can already do with the worker / push_worker commands?

[untitaker]

What do you mean by this? Isn't that what they can already do with the worker / push_worker commands?

right now there are two entrypoints for taskworker CLI that can be mounted in the application's CLI in different places. it is permitted to do this, though a bad idea:

sentry run worker
sentry push-worker

it would be more restrictive, and less boilerplate in applcation code if instead only one CLI needs to be mounted.

sentry run worker
sentry run push-worker

could be:

sentry run worker --push-mode

or:

sentry run worker push
sentry run worker pull

in the ops repo template, one then only has to configure sentry run worker as "the prefix where the worker CLI lives", and everything else can be done using shared templates

[evanh]

@untitaker I think I see your point, but I deliberately separated them because they require different arguments. If I had one command worker, that command would have to take the union between all the different CLI args, and it would be unclear which ones were needed for which mode. What do you think?

[untitaker]

I deliberately separated them because they require different arguments

yeah i see that. but you can make them sub-commands of a @click.group (in this repo), which then would be added to another click group in the app (sentry run worker pull and sentry run worker push)

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit bf572dd. Configure here.}

[cursor]

Track futures flag overwrites env

Medium Severity

Both pull and push always assign ARROYO_TRACK_PRODUCER_FUTURES from the Click flag, whose default is false. Deployments that enable tracking only via the environment lose that setting when the new worker commands run without --track-producer-futures, unlike --grpc-max-message-size, which leaves the env var in effect when unset.

Additional Locations (1)

• clients/python/src/taskbroker_client/cli.py#L364-L365

 

^{Reviewed by Cursor Bugbot for commit bf572dd. Configure here.}

[untitaker]

i think spawn and scheduler can be in that group too, i forgot to mention them but they are also distinct entrypoints that need separate wiring up rn

[sentry]

Bug: The pull command worker incorrectly re-raises KeyboardInterrupt on graceful shutdown, causing a non-zero exit code instead of a clean exit with code 0.
_{Severity: HIGH}

Suggested Fix

Modify TaskWorker.start() to catch the KeyboardInterrupt and allow the function to complete and return an exit code, similar to how PushTaskWorker.start() is implemented. This will prevent the exception from propagating and ensure the CLI can call SystemExit(0) for a clean shutdown.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: clients/python/src/taskbroker_client/cli.py#L210-L211

Potential issue: When the `pull` command worker receives a `SIGINT` or `SIGTERM` signal
for a graceful shutdown, the underlying `TaskWorker.start()` method re-raises a
`KeyboardInterrupt` after handling the signal. This unhandled exception propagates up,
preventing the CLI from executing `raise SystemExit(exitcode)`. Instead, the process
terminates with a non-zero exit code (e.g., 130), incorrectly signaling a crash to
orchestration systems like Kubernetes. The `push` command worker correctly swallows this
exception and exits cleanly with code 0.