surface error from the controller async tasks

[felipemello1]

• I have personally reviewed this PR and description before asking others to do so. It meets the quality bar I expect from others. I understand that if this PR is perceived as unverified AI-generated code, it will be closed without further explanation.
• I have run tests and confirmed that this code works

---

Description

When an exception occurs in a background task (e.g., continuous_rollouts), two issues prevented clean error handling:

1. Exceptions were silently swallowed: asyncio.create_task() fire-and-forget tasks don't propagate exceptions to the main process. The program just froze.

2. Training loop ignored shutdown: continuous_training only checked max_steps, not shutdown_event.is_set(). Even when rollouts crashed and set the shutdown event, training kept running.

Solution

1. Add on_task_done callback to surface background task exceptions and trigger shutdown
2. Add shutdown_event.is_set() check to continuous_training loop
3. Catch Exception (not just KeyboardInterrupt) on the awaited training task

Test plan

1. added a bug
2. ran without the changes -> it just froze
3. ran with the changes -> got

Traceback (most recent call last):
  File "/home/felipemello/forge/apps/grpo/main.py", line 150, in continuous_rollouts
    raise ValueError("DELIBERATE BUG: Testing exception visibility!")
ValueError: DELIBERATE BUG: Testing exception visibility!
[...]
[actor=<root>] Shutting down provisioner..
[actor=<root>] Shutting down 2 service(s) and 4 actor(s)...
[actor=<root>] Health loop stopped gracefully.

[JenniferWang]

This is to surface error from the controller async tasks. Maybe use a better title to indicate the scope.
Followup: what's the behavior when exception happens in remote actors? Do you have visibility from monarch to help you see that?

[felipemello1]

what's the behavior when exception happens in remote actors?

good question, i think i have to add

from monarch.config import configure
configure(enable_log_forwarding=True, enable_file_capture=True)

i will test it later and make a PR

[felipemello1]

@JenniferWang , it does surface errors in monarch actors

monarch._src.actor.actor_mesh.ActorError: Actor call TitanTrainer.train_step failed.
 Traceback of where the remote call failed (most recent call last):
  File "/home/felipemello/.conda/envs/forge/lib/python3.12/site-packages/monarch/_src/actor/actor_mesh.py", line 1208, in handle
    result = await the_method(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/felipemello/forge/src/forge/actors/trainer/titan.py", line 161, in train_step
    raise ValueError("DELIBERATE BUG IN ACTOR: Testing Monarch error visibility!")
ValueError: DELIBERATE BUG IN ACTOR: Testing Monarch error visibility!