{ "cells": [ { "cell_type": "markdown", "id": "9846dce9", "metadata": {}, "source": [ "---\n", "title: \"Run a managed workflow\"\n", "description: \"Run a managed workflow, check its status, result, and logs, find the cause of an error, and cancel a job.\"\n", "---\n", "\n", "{/* cspell:ignore maxwait squeue scancel */}\n", "\n", "# Run a managed workflow\n", "\n" ] }, { "cell_type": "markdown", "id": "133c531e", "metadata": { "tags": [ "version-info" ] }, "source": [ "{/*\n", " DO NOT EDIT THIS CELL!!!\n", " This cell's content is generated automatically by a script. Anything you add\n", " here will be removed next time the notebook is run. To add new content, create\n", " a new cell before or after this one.\n", " */}\n", "\n", "\n", " \n", " The code on this page was developed using the following requirements.\n", " We recommend using these versions or newer.\n", "\n", " ```\n", " qiskit-serverless~=0.38.0\n", " ```\n", " \n", "\n", "\n" ] }, { "cell_type": "markdown", "id": "ea01528c", "metadata": {}, "source": [ "A managed workflow job works like a batch job that you submit to an HPC scheduler. `run()` submits the job and returns a `Job` object immediately. Use the `Job` object for the tasks that `squeue`, the job's output file, and `scancel` cover on a cluster. The job keeps running after you close your notebook.\n", "\n", "The examples on this page use a small workflow that reads one angle, `theta`, and fails if the angle is out of range. Upload it as described in [Deploy your first managed workflow](/docs/managed-workflows/deploy-first-managed-workflow).\n", "\n" ] }, { "cell_type": "code", "execution_count": 2, "id": "4c3fd41c", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Writing ./source_files/my_workflow.py\n" ] } ], "source": [ "%%writefile ./source_files/my_workflow.py\n", "import math\n", "\n", "from qiskit_serverless import get_arguments, save_result\n", "\n", "arguments = get_arguments()\n", "theta = arguments[\"theta\"]\n", "if not 0 <= theta <= math.pi:\n", " raise ValueError(f\"theta must be between 0 and pi, got {theta}\")\n", "\n", "print(f\"Computing the expectation value for theta={theta}\", flush=True)\n", "save_result({\"theta\": theta, \"expectation_value\": math.cos(theta)})" ] }, { "cell_type": "markdown", "id": "99018df7", "metadata": {}, "source": [ "## Start a job\n", "\n", "Get the workflow by its title, and call `run()` with the arguments that the workflow reads:\n", "\n" ] }, { "cell_type": "markdown", "id": "0928b066", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 4, "id": "6c7b19db", "metadata": {}, "outputs": [], "source": [ "from qiskit_ibm_catalog import ManagedWorkflowClient\n", "\n", "client = ManagedWorkflowClient()\n", "my_workflow = client.load(\"my-workflow\")\n", "\n", "job = my_workflow.run(theta=0.5)\n", "job_id = job.job_id" ] }, { "cell_type": "markdown", "id": "afe4cf05", "metadata": {}, "source": [ "`run()` returns after the workflow queues the job. Save the job ID so that you can [find the job from a new session](#find-a-job-from-a-new-session).\n", "\n", "* To choose which arguments a workflow accepts and check them before the job starts, see [Define inputs and outputs](/docs/managed-workflows/managed-workflow-inputs-and-outputs).\n", "* To run the job on a different compute size, pass `function_size`. See [Choose a size for each run](/docs/managed-workflows/managed-workflow-compute-profiles#choose-a-size-for-each-run).\n", "\n" ] }, { "cell_type": "markdown", "id": "ba179ad7", "metadata": {}, "source": [ "## Follow a job\n", "\n", "### Check the status\n", "\n", "Immediately after you call `run()`, the status is usually `QUEUED`:\n", "\n" ] }, { "cell_type": "markdown", "id": "ad6b65f6", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 5, "id": "421e1233", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "'QUEUED'" ] }, "execution_count": 5, "metadata": {}, "output_type": "execute_result" } ], "source": [ "job.status()" ] }, { "cell_type": "markdown", "id": "dd579097", "metadata": {}, "source": [ "| Status | Meaning |\n", "| ------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |\n", "| `QUEUED` | The job waits for a compute slot. |\n", "| `INITIALIZING` | The workflow is starting an environment with the job's compute profile. |\n", "| `RUNNING` | The workflow's entrypoint is running. |\n", "| `RUNNING: ` | The entrypoint reported a stage with [`update_status()`](/docs/managed-workflows/manage-quantum-classical-workflows#report-progress-with-update_status), such as `RUNNING: WAITING_FOR_QPU`. |\n", "| `CANCELING` | You canceled the job, and the workflow is stopping it. |\n", "| `DONE` | The entrypoint exited successfully. |\n", "| `ERROR` | The entrypoint raised an error or exited with a nonzero code. See [Find the cause of an error](#find-the-cause-of-an-error). |\n", "| `CANCELED` | You canceled the job, or it reached the [24-hour limit](#time-limit). |\n", "\n", "`DONE`, `ERROR`, and `CANCELED` are final, and `job.in_terminal_state()` returns `True` for them. A status loop that waits for another value, such as `SUCCEEDED`, never ends.\n", "\n", "#### Wait for the job to end\n", "\n", "To wait until the job ends, call `wait_for_final_state()`. It returns the final status, and with `verbose=True` it prints each status change. If you set `timeout` in seconds, `wait_for_final_state()` raises a `TimeoutError` when the time runs out before the job ends. The job keeps running:\n", "\n", "\n", "\n", "```python\n", "job.wait_for_final_state(timeout=600, verbose=True)\n", "```\n", "\n", "```text\n", "Waiting for job 3f2a1c0e-6b1d-4c0e-9a51-2f0c4d1e7a90\n", "0:00:00 QUEUED\n", "0:02:10 RUNNING\n", "0:03:05 RUNNING: WAITING_FOR_QPU\n", "0:41:44 DONE\n", "'DONE'\n", "```\n", "\n", "`wait_for_final_state()` checks the status after 1, 2, 4, 8, 16, and 32 seconds, and then every 60 seconds, so it sees the end of a short job within seconds and makes few requests for a long job. To check at a fixed interval instead, pass `poll_interval` in seconds.\n", "\n" ] }, { "cell_type": "markdown", "id": "feab6243", "metadata": {}, "source": [ "### Read the logs\n", "\n", "Anything that the workflow writes to standard output or standard error goes to the job's logs. You can read them with `job.logs()` while the job runs, before it has a result:\n", "\n" ] }, { "cell_type": "markdown", "id": "075be982", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 7, "id": "858d8221", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Computing the expectation value for theta=0.5\n", "\n" ] } ], "source": [ "print(job.logs())" ] }, { "cell_type": "markdown", "id": "9d544e77", "metadata": {}, "source": [ "* The workflow uploads new log lines about every 15 seconds, and one last time when the job ends or is canceled. Immediately after a job starts, `job.logs()` returns `No logs yet.`\n", "* The workflow keeps the newest 50 MB of logs for each job, and adds a line at the top that says the logs were truncated.\n", "\n", "To decide what your workflow logs, see [Write logs](/docs/managed-workflows/managed-workflow-logs-and-errors#write-logs). To show only the lines that match a regular expression, pass `include` to `job.filtered_logs()`. To hide lines, pass `exclude`.\n", "\n" ] }, { "cell_type": "markdown", "id": "5a0fea10", "metadata": {}, "source": [ "### Get the result\n", "\n", "`job.result()` waits until the job ends, then returns the value that the workflow passed to [`save_result()`](/docs/managed-workflows/managed-workflow-inputs-and-outputs#save-results):\n", "\n" ] }, { "cell_type": "markdown", "id": "89ccb364", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 9, "id": "0fefeb2c", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "{'theta': 0.5, 'expectation_value': 0.8775825618903728}" ] }, "execution_count": 9, "metadata": {}, "output_type": "execute_result" } ], "source": [ "job.result()" ] }, { "cell_type": "markdown", "id": "5e61d330", "metadata": {}, "source": [ "By default, `job.result()` checks the status every 30 seconds and waits with no time limit. Pass `maxwait` to limit the number of checks. For example, `job.result(maxwait=40)` waits about 20 minutes. To check one time without waiting, pass `wait=False`. If `job.result()` stops waiting before the job ends, it returns `{}`.\n", "\n" ] }, { "cell_type": "markdown", "id": "ce86e0db", "metadata": {}, "source": [ "### Check how long each stage took\n", "\n", "`job.events()` lists every status and stage change of a job with its time, so you can see how long the job waited for compute and how long each stage took:\n", "\n" ] }, { "cell_type": "markdown", "id": "0803cb89", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 11, "id": "aad1d233", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "2026-11-02T14:43:07.134377Z QUEUED\n", "2026-11-02T14:43:24.474199Z PENDING\n", "2026-11-02T14:46:07.309519Z RUNNING\n", "2026-11-02T14:46:21.115038Z SUCCEEDED\n" ] } ], "source": [ "for event in reversed(job.events()):\n", " print(\n", " event.created,\n", " event.data.get(\"status\") or event.data.get(\"sub_status\"),\n", " )" ] }, { "cell_type": "markdown", "id": "d769cd8a", "metadata": {}, "source": [ "The events use the stored status names. `PENDING` is `INITIALIZING`, and `SUCCEEDED` is `DONE`. For the QPU usage of a run, see [Find the QPU jobs of a run](/docs/managed-workflows/manage-quantum-classical-workflows#find-the-qpu-jobs-of-a-run).\n", "\n" ] }, { "cell_type": "markdown", "id": "c6e0250d", "metadata": {}, "source": [ "## Find the cause of an error\n", "\n", "When the workflow raises an exception, the job ends with the `ERROR` status. For example, run the workflow with an angle that it rejects:\n", "\n" ] }, { "cell_type": "markdown", "id": "e7bba7b9", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 13, "id": "4e0a6768", "metadata": {}, "outputs": [], "source": [ "failed_job = my_workflow.run(theta=-1.0)" ] }, { "cell_type": "markdown", "id": "0336679e", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 15, "id": "85c646a3", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "'ERROR'" ] }, "execution_count": 15, "metadata": {}, "output_type": "execute_result" } ], "source": [ "failed_job.status()" ] }, { "cell_type": "markdown", "id": "ba1b8214", "metadata": {}, "source": [ "The logs show the traceback:\n", "\n" ] }, { "cell_type": "markdown", "id": "d6328994", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 16, "id": "e2a27234", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Traceback (most recent call last):\n", " File \"/function_user_data/my_workflow.py\", line 8, in \n", " raise ValueError(f\"theta must be between 0 and pi, got {theta}\")\n", "ValueError: theta must be between 0 and pi, got -1.0\n", "\n" ] } ], "source": [ "print(failed_job.logs())" ] }, { "cell_type": "markdown", "id": "2ca02fe9", "metadata": {}, "source": [ "`job.result()` raises a `QiskitServerlessException`:\n", "\n" ] }, { "cell_type": "markdown", "id": "3a2ed6bc", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 17, "id": "aa76e1b0", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ " raise ValueError(f\"theta must be between 0 and pi, got {theta}\")\n", "ValueError: theta must be between 0 and pi, got -1.0\n", "\n" ] } ], "source": [ "from qiskit_serverless.exception import QiskitServerlessException\n", "\n", "try:\n", " failed_job.result()\n", "except QiskitServerlessException as error:\n", " print(error)" ] }, { "cell_type": "markdown", "id": "6fe30140", "metadata": {}, "source": [ "If the workflow [reports errors with `send_error()`](/docs/managed-workflows/managed-workflow-logs-and-errors#best-practice-report-errors-with-send_error), read them with `job.events(type=\"ERROR\")`, and read their messages with `job.error_message()`. For a job without reported errors, `job.events(type=\"ERROR\")` returns an empty list, and `job.error_message()` returns `{}`.\n", "\n", "{/* REVIEW: for custom workflows on Fleets, a raised ServerlessError is not recorded as an error event: only provider jobs run inside gateway/templates/main.tmpl, which calls send_error(). Custom jobs run the entrypoint directly through fleet_custom_job_wrapper.py. Confirm whether custom jobs should get the same handler, and whether error_message() should return a string instead of {}. */}\n", "\n", "\n", "\n", "```python\n", "for event in job.events(type=\"ERROR\"):\n", " print(event.created, event.data.get(\"code\"), event.data.get(\"message\"))\n", "```\n", "\n", "## Manage jobs\n", "\n", "### Cancel a job\n", "\n", "Cancel a job with `job.cancel()`. The status changes to `CANCELING` while the workflow stops the job, and then to `CANCELED`:\n", "\n" ] }, { "cell_type": "markdown", "id": "8abb82c8", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 19, "id": "fb36d850", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "'Job has been stopped. No active runtime job ID associated with this serverless job ID.'" ] }, "execution_count": 19, "metadata": {}, "output_type": "execute_result" } ], "source": [ "job_to_cancel = my_workflow.run(theta=1.0)\n", "job_to_cancel.cancel()" ] }, { "cell_type": "markdown", "id": "5ee343cb", "metadata": {}, "source": [ "`job.cancel()` returns a message about the job and about its [linked QPU jobs](/docs/managed-workflows/manage-quantum-classical-workflows#cancel-linked-qpu-jobs), which it also tries to cancel. For a running job, the workflow sends `SIGTERM` to your entrypoint. To save partial results, see [Clean up when a job is canceled](/docs/managed-workflows/managed-workflow-logs-and-errors#clean-up-when-a-job-is-canceled). If the job already ended, `job.cancel()` returns `Job already in terminal state.` and changes nothing.\n", "\n", "{/* REVIEW: in a live test (1 Nov 2026), a job canceled while RUNNING went to CANCELING and then DONE, with its result saved, because the entrypoint finished before the stop took effect. Confirm whether a job that finishes during CANCELING should end as DONE or CANCELED. */}\n", "\n", "### Find a job from a new session\n", "\n", "The `Job` object and the workflow exist only in your current Python session. After you restart a notebook kernel, pass the saved job ID to `client.job()`, or list your recent jobs with `client.jobs()`:\n", "\n" ] }, { "cell_type": "markdown", "id": "8beaddc3", "metadata": {}, "source": [ "\n", "\n" ] }, { "cell_type": "code", "execution_count": 21, "id": "ddd987e4", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "[,\n", " ,\n", " ,\n", " ,\n", " ]" ] }, "execution_count": 21, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from qiskit_ibm_catalog import ManagedWorkflowClient\n", "\n", "client = ManagedWorkflowClient()\n", "job = client.job(job_id)\n", "my_workflow = client.load(\"my-workflow\")\n", "\n", "client.jobs(function=my_workflow, status=\"ERROR\", limit=5)" ] }, { "cell_type": "markdown", "id": "38c4efcd", "metadata": {}, "source": [ "`client.jobs()` returns your 10 most recent jobs, newest first. Filter it with `function`, `status` (from the [status table](#check-the-status)), or `created_after` (an ISO 8601 timestamp), and page through it with `limit` and `offset`. To list the workflows that you uploaded, call `client.list()`.\n", "\n", "The [Workloads](/workloads) table on IBM Quantum Platform also lists your jobs, with their status, compute profile, and run times.\n", "\n" ] }, { "cell_type": "markdown", "id": "a461942f", "metadata": {}, "source": [ "### Time limit\n", "\n", "A managed workflow cancels a job that runs for more than 24 hours, and the job ends with the `CANCELED` status. During long runs, [save checkpoints to file storage](/docs/managed-workflows/manage-files-across-managed-workflows#best-practice-checkpoint-long-jobs) so that you do not lose work.\n", "\n", "{/* REVIEW: the timeout is measured from the job's latest event, not from its start (gateway/scheduler/tasks/update_fleets_jobs_statuses.py stop_job_if_timeout). Calls to update_status() and send_error() create events, so a job that reports stages can run longer than 24 hours. This looks like a bug; confirm the intended behavior. */}\n", "\n", "## Troubleshoot jobs\n", "\n", "| Problem | Cause |\n", "| ------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |\n", "| `run()` fails with `Code: 429` and `Active job limit reached.` | You reached the [active job limit](/docs/managed-workflows/run-parameter-sweep#active-job-limit). Wait for some jobs to end, or [cancel](#cancel-a-job) jobs that you no longer need. |\n", "| The job stays `QUEUED`, and `run()` warned that `your account, plan or subscription is blocked`. | The job does not start until the block on your account, plan, or subscription is lifted. Contact your administrator. |\n", "| `job.logs()` returns `No logs yet.` | The job is queued or just started. Wait about 15 seconds. |\n", "| `job.result()` returns `{}` | The job has not ended, was canceled, or saved no result. See [Save results](/docs/managed-workflows/managed-workflow-inputs-and-outputs#save-results). |\n", "| `job.error_message()` returns `{}` | The workflow raised an exception without reporting an error. Read the traceback with `job.logs()`. |\n", "| The job is `CANCELED`, but you did not cancel it | The job reached the [24-hour limit](#time-limit). |\n", "\n", "## Next steps\n", "\n", "\n", " * [Write logs and handle errors](/docs/managed-workflows/managed-workflow-logs-and-errors) to set up the logs and errors that a job reports.\n", " * [Run a parameter sweep](/docs/managed-workflows/run-parameter-sweep) to run the workflow one time for each set of inputs.\n", "\n", "\n" ] }, { "cell_type": "markdown", "metadata": {}, "id": "a1b8767d", "source": "© IBM Corp., 2017-2026" } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3" } }, "nbformat": 4, "nbformat_minor": 5 }