Errors & debugging

The exception hierarchy, retry introspection, and interactive debugging.

Exception hierarchy

All SDK exceptions inherit from GworkerError. Catch specific subclasses for structured error handling, or catch GworkerError as a fallback.

python
1import gworker_client as gw2 3try:4    result = my_task.run(x)5except gw.TimeoutError:6    # task exceeded its timeout= limit7    log.warning("timed out, retrying later")8except gw.ExecutionError as e:9    # task body raised a Python exception10    log.error("task failed: %s", e.__cause__)11except gw.RemoteError as e:12    # non-zero exit code or structured remote failure13    log.error("remote error %d: %s", e.exit_code, e.stderr)14except gw.ServiceError:15    # gworker service is degraded — retry with backoff16    raise

Tail a failing run

Stream stderr from all workers. The --env flag selects the stage.

bash
1gworker logs my-app --env dev --follow

Interactive debug pause

Call gw.interact() inside a task to pause execution and request an interactive shell. Connect with `gworker shell` from a separate terminal. Execution resumes after you exit.

python
1@app.task(gpu=gw.Gpu("A100"))2async def train(config: dict):3    model = build_model(config)4    loss = first_forward_pass(model)5    gw.interact()  # pause here — inspect model, loss, GPU state6    continue_training(model)

Attempt introspection

Read the current retry attempt number inside the task body to implement attempt-aware recovery logic.

python
1@app.task(retries=3)2async def process(item_id: str):3    attempt = gw.current_attempt_number()4    if attempt > 1:5        cleanup_partial_work(item_id)6    do_work(item_id)

Suppress noisy output

Use gw.enable_output as a context manager to suppress stdout/stderr from a specific code block without silencing the rest of the task.

python
1@app.task()2async def run():3    with gw.enable_output(False):4        verbose_library_call()  # stdout suppressed5    log.info("done")  # this still appears in logs

Local mode guard

Use gw.is_local() to branch between local dev and remote execution. Prevents GPU-only code from crashing in unit tests.

python
1@app.task(gpu=gw.Gpu("A100"))2async def infer(text: str) -> str:3    if gw.is_local():4        return f"[local stub] {text}"5    import torch6    return run_on_gpu(text)