diff --git a/docs/contributor_guide/event-bus.md b/docs/contributor_guide/event-bus.md
index 021f42c6a0b..885a2afcac1 100644
--- a/docs/contributor_guide/event-bus.md
+++ b/docs/contributor_guide/event-bus.md
@@ -193,6 +193,31 @@ accurate conventions, add the proposed contract to the registry in the same cont
 treat maintainer review as establishing the standard for future implementations of that operation.
 Do not create an undocumented vendor-local alias for an existing concept.
 
+### Manual operator actions
+
+Manual operations use `manual_operator.<action>` so the event records both the manual executor and
+the semantic work requested. Represent the `ManualOperator` as `device`, list any direct modeled
+resources in `resources`, and preserve action-specific request data without substituting inferred
+deck resources. When an automated counterpart defines canonical parameter names and units, reuse
+them inside the manual operation's `details`. A genuine manual resource transfer additionally
+includes its actual `source` and `destination` resource references.
+
+```python
+{
+  "device": device_reference(manual_operator, name=manual_operator.name),
+  "resources": [resource_reference(plate)],
+  "manual_action": "centrifuge.spin",
+  "title": "Spin sample plate",
+  "details": {
+    "relative_centrifugal_force": 300,
+    "duration": 180,
+  },
+}
+```
+
+Operator cancellation or a provider-reported failure is a failed lifecycle outcome. Completion
+metadata such as `confirmed_by` belongs only on the `.completed` event.
+
 ## Failure events
 
 A failed operation retains the original invocation context and adds:
diff --git a/docs/contributor_guide/event-schemas.md b/docs/contributor_guide/event-schemas.md
index 337d48d9d52..5f7388541c3 100644
--- a/docs/contributor_guide/event-schemas.md
+++ b/docs/contributor_guide/event-schemas.md
@@ -165,6 +165,21 @@ These are state-transition records rather than semantic operation lifecycles.
 | `liquid_handler.resource_move` | `device`, `resources` | Moves the currently held resource without assigning it to a destination. |
 | `liquid_handler.resource_drop` | `device`, `resources`, `destination` | Drops the currently held resource at a resource or geometric destination. |
 
+### Manual operator actions
+
+Manual actions use the semantic lifecycle `manual_operator.<action>.*`, where `<action>` is a
+stable, developer-defined action identifier such as `centrifuge.spin`, `plate_reader.read`, or
+`quality_control.inspect`.
+
+| Operation | Fields | Notes |
+| --- | --- | --- |
+| `manual_operator.<action>` | `device`, optional `resources`, `manual_action`, `title`, `instructions`, `confirmation_text`, `details`; **completed only:** optional `confirmed_by`, optional `result_message` | `device` is the `ManualOperator`; `details` contains action-specific request data. When the action has an automated counterpart, reuse its canonical field names and PLR default units inside `details`. |
+| `manual_operator.resource.move` | `device`, `resources`, `source`, `destination`, `manual_action`, `title`, `instructions`, `confirmation_text`, optional `details`; **completed only:** optional `confirmed_by`, optional `result_message` | `resources` contains the directly moved resource. `source` and `destination` are its actual modeled transfer endpoints. When supplied, `details.destination_rotation` is the explicit local rotation relative to `destination`, not an absolute/world rotation. PLR composes its resulting absolute rotation with the destination's absolute rotation; use the local pose that an equivalent automated transfer would produce. It is never inferred from the destination. The subsequent model update emits normal `resource.unassigned` and `resource.assigned` state transitions. |
+
+Manual action providers decide how an operator acknowledges the request. Cancellation,
+provider-reported failure, invalid provider results, and provider exceptions produce the normal
+failed lifecycle record with `error_type` and `error_message`.
+
 ## Liquid handling
 
 ### Channelized liquid operations
diff --git a/docs/cookbook/index.rst b/docs/cookbook/index.rst
index 1217efbebf2..df08b560c21 100644
--- a/docs/cookbook/index.rst
+++ b/docs/cookbook/index.rst
@@ -38,6 +38,15 @@ teach, and accelerate your own automation workflows.
    :link: slack_notifications.html
    :tags: Notifications Slack Monitoring EventBus
 
+.. plrcard::
+   :header: Use ManualOperator in a Jupyter notebook
+   :card_description: <ul>
+      <li>Pause a notebook for an acknowledged manual handoff</li>
+      <li>Reconcile a manually moved plate in PLR's resource model</li>
+      <li>Observe incubator and ManualOperator lifecycle events</li></ul>
+   :link: manual_operator_jupyter.html
+   :tags: ResourceMovement EventBus
+
 .. plrcardgrid::
 
 .. End of tutorial card section
@@ -51,3 +60,4 @@ teach, and accelerate your own automation workflows.
 
    star_movement_plate_to_alpaqua_core
    slack_notifications
+   manual_operator_jupyter
diff --git a/docs/cookbook/manual_operator_jupyter.ipynb b/docs/cookbook/manual_operator_jupyter.ipynb
new file mode 100644
index 00000000000..2925733aa9b
--- /dev/null
+++ b/docs/cookbook/manual_operator_jupyter.ipynb
@@ -0,0 +1,250 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Manual operator actions in a Jupyter notebook\n"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "This recipe models a common hybrid workflow: a plate is fetched from an incubator, a person moves it into a plate reader, and the protocol then reads it.\n",
+    "\n",
+    "`ManualOperator` keeps the protocol independent of the acknowledgement interface. This notebook uses a small local provider built on `input()` because a notebook prompt is often the right level of complexity for a simple manual handoff. The same `ManualOperator` calls can later use a dashboard, LIMS, or message-broker provider without changing the protocol logic.\n"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Prerequisites\n",
+    "\n",
+    "- PyLabRobot with the `manual_operator` and EventBus features available.\n",
+    "- This recipe uses legacy chatterbox backends, so it does **not** connect to hardware.\n",
+    "- Run the notebook interactively. The manual-transfer cell pauses until the operator presses Enter. Set `INTERACTIVE = False` to run the chatterbox demonstration without a prompt.\n"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 1. Define a notebook-local provider\n",
+    "\n",
+    "An `OperatorActionProvider` turns a transport-independent `OperatorActionRequest` into an acknowledgement interaction. This minimal provider prints the request and treats Enter as a successful acknowledgement.\n",
+    "\n",
+    "It intentionally pauses this notebook's event loop while waiting. That is appropriate when the protocol should wait for the manual handoff before proceeding. Applications that need a richer UI or concurrent orchestration can supply their own provider instead.\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pylabrobot.manual_operator import (\n",
+    "    ManualOperator,\n",
+    "    OperatorActionRequest,\n",
+    "    OperatorActionResult,\n",
+    ")\n",
+    "\n",
+    "\n",
+    "INTERACTIVE = True\n",
+    "\n",
+    "\n",
+    "class NotebookOperatorActionProvider:\n",
+    "    \"\"\"Minimal Jupyter-friendly provider for interactive protocol pauses.\"\"\"\n",
+    "\n",
+    "    async def request(self, action: OperatorActionRequest) -> OperatorActionResult:\n",
+    "        print(f\"\\n{action.title}\\n\\n{action.instructions}\\n\")\n",
+    "        if INTERACTIVE:\n",
+    "            input(f\"{action.confirmation_text}: \")\n",
+    "        else:\n",
+    "            print(f\"[auto-confirmed] {action.confirmation_text}\")\n",
+    "        return OperatorActionResult.completed(confirmed_by=\"notebook operator\")\n",
+    "\n",
+    "\n",
+    "operator = ManualOperator(NotebookOperatorActionProvider(), name=\"notebook_operator\")\n"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 2. Model the incubator, plate reader, and sample plate\n",
+    "\n",
+    "The sample plate begins in a modeled incubator storage site. The incubator and reader use chatterbox backends, which print their actions and return deterministic dummy data.\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pylabrobot.events import EventBus, PLREvent, use_event_bus\n",
+    "from pylabrobot.legacy.plate_reading import PlateReader, PlateReaderChatterboxBackend\n",
+    "from pylabrobot.legacy.storage import Incubator, IncubatorChatterboxBackend\n",
+    "from pylabrobot.resources import Coordinate, PlateCarrier, PlateHolder\n",
+    "from pylabrobot.resources.corning import cor_96_wellplate_360uL_Fb\n",
+    "\n",
+    "\n",
+    "incubator_slot = PlateHolder(\n",
+    "    name=\"incubator_slot_1\",\n",
+    "    size_x=127.76,\n",
+    "    size_y=85.48,\n",
+    "    size_z=20,\n",
+    "    pedestal_size_z=0,\n",
+    ").at(Coordinate.zero())\n",
+    "incubator_rack = PlateCarrier(\n",
+    "    name=\"incubator_rack\",\n",
+    "    size_x=140,\n",
+    "    size_y=100,\n",
+    "    size_z=100,\n",
+    "    sites={0: incubator_slot},\n",
+    ")\n",
+    "\n",
+    "incubator = Incubator(\n",
+    "    name=\"incubator\",\n",
+    "    size_x=200,\n",
+    "    size_y=200,\n",
+    "    size_z=300,\n",
+    "    backend=IncubatorChatterboxBackend(),\n",
+    "    racks=[incubator_rack],\n",
+    "    loading_tray_location=Coordinate.zero(),\n",
+    ")\n",
+    "plate_reader = PlateReader(\n",
+    "    name=\"plate_reader\",\n",
+    "    size_x=160,\n",
+    "    size_y=160,\n",
+    "    size_z=100,\n",
+    "    backend=PlateReaderChatterboxBackend(),\n",
+    ")\n",
+    "\n",
+    "sample_plate = cor_96_wellplate_360uL_Fb(name=\"sample_plate\")\n",
+    "incubator_slot.assign_child_resource(sample_plate)\n",
+    "\n",
+    "print(f\"{sample_plate.name} starts in {incubator_slot.name}.\")\n"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 3. Optional: observe semantic EventBus events\n",
+    "\n",
+    "`ManualOperator` does not require EventBus. It works with no subscriber installed. This optional section demonstrates that the incubator fetch and manual resource transfer emit semantic lifecycle events when a subscriber is active. Set `ENABLE_EVENT_BUS_DEMO = False` to run the exact same manual workflow without EventBus.\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from contextlib import nullcontext\n",
+    "\n",
+    "\n",
+    "ENABLE_EVENT_BUS_DEMO = True\n",
+    "\n",
+    "event_bus = EventBus()\n",
+    "\n",
+    "\n",
+    "def print_operation_outcome(event: PLREvent) -> None:\n",
+    "    operation = event.context.get(\"operation\")\n",
+    "    if not isinstance(operation, str):\n",
+    "        return\n",
+    "    outcome = event.name.removeprefix(f\"{operation}.\")\n",
+    "    if outcome in {\"completed\", \"failed\"}:\n",
+    "        print(f\"[event] {event.name}\")\n",
+    "\n",
+    "\n",
+    "if ENABLE_EVENT_BUS_DEMO:\n",
+    "    event_bus.subscribe(print_operation_outcome)\n"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 4. Fetch, manually transfer, and read the plate\n",
+    "\n",
+    "The model remains unchanged while the request is pending: the sample plate stays on the incubator tray until the operator reports completion. `move_resource()` then validates the model and assigns the plate to the reader. On real hardware, ensure the reader is open before moving the plate and close it before reading. The `ENABLE_EVENT_BUS_DEMO` setting changes only observation; it does not change the manual action or resource-model behavior.\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "async def run_manual_transfer_and_read() -> list[dict]:\n",
+    "    event_scope = use_event_bus(event_bus) if ENABLE_EVENT_BUS_DEMO else nullcontext()\n",
+    "    with event_scope:\n",
+    "        await incubator.setup()\n",
+    "        await plate_reader.setup()\n",
+    "        try:\n",
+    "            await incubator.fetch_plate_to_loading_tray(sample_plate.name)\n",
+    "            assert incubator.loading_tray.resource is sample_plate\n",
+    "\n",
+    "            await plate_reader.open()\n",
+    "            await operator.move_resource(\n",
+    "                resource=sample_plate,\n",
+    "                source=incubator.loading_tray,\n",
+    "                destination=plate_reader,\n",
+    "                title=\"Move plate to reader\",\n",
+    "                instructions=(\n",
+    "                    \"Move sample_plate from the incubator loading tray into the open plate reader, \"\n",
+    "                    \"and confirm after it is seated correctly.\"\n",
+    "                ),\n",
+    "                confirmation_text=\"Press Enter after the plate is seated in the reader\",\n",
+    "                details={\"reason\": \"manual incubator-to-reader handoff\"},\n",
+    "            )\n",
+    "            assert plate_reader.get_plate() is sample_plate\n",
+    "\n",
+    "            # Close the reader only after the model is updated to show the plate inside it.\n",
+    "            await plate_reader.close()\n",
+    "            return await plate_reader.read_absorbance(\n",
+    "                wavelength=450,\n",
+    "                use_new_return_type=True,\n",
+    "            )\n",
+    "        finally:\n",
+    "            await plate_reader.stop()\n",
+    "            await incubator.stop()\n",
+    "\n",
+    "\n",
+    "readings = await run_manual_transfer_and_read()\n",
+    "print(readings[0][\"data\"][0][:3])\n"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## What the protocol guarantees\n",
+    "\n",
+    "- The incubator fetch updates the model from its storage site to the loading tray.\n",
+    "- A cancelled or failed manual action leaves the plate on the tray in the PLR model.\n",
+    "- A successful `move_resource()` acknowledgement updates the model only if the source and destination are still consistent.\n",
+    "- If an EventBus subscriber is active, it sees the incubator fetch and the `manual_operator.resource.move` lifecycle. No subscriber is required for the manual action or resource-model update to work.\n",
+    "\n",
+    "For manual actions that do not move a modeled resource, call `await operator.perform(...)` with a stable action name and structured `details` instead.\n"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "name": "python",
+   "version": "3"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}
diff --git a/docs/user_guide/index.md b/docs/user_guide/index.md
index 8aeff3139df..611aef2d803 100644
--- a/docs/user_guide/index.md
+++ b/docs/user_guide/index.md
@@ -61,6 +61,7 @@ machine-agnostic-features/tip-spot-generators
 machine-agnostic-features/logging-and-validation/logging-and-validation
 machine-agnostic-features/error-handling-general
 machine-agnostic-features/sila-discovery
+machine-agnostic-features/manual-operator
 ```
 
 ```{toctree}
diff --git a/docs/user_guide/machine-agnostic-features/event-bus.md b/docs/user_guide/machine-agnostic-features/event-bus.md
index 59d484df5f1..8b9fcabb94c 100644
--- a/docs/user_guide/machine-agnostic-features/event-bus.md
+++ b/docs/user_guide/machine-agnostic-features/event-bus.md
@@ -126,6 +126,7 @@ events.
 | `agilent.vspin.VSpin` | `centrifuge.spin` |
 | `agilent.vspin.Access2` | `centrifuge_loader.load`, `centrifuge_loader.unload` |
 | `brooks.precise_flex.PreciseFlex` | lifecycle, fault/home/freedrive, joint/cartesian/rail/gripper motion, pick/drop, park |
+| `manual_operator.ManualOperator` | arbitrary acknowledged manual actions; resource moves |
 
 Detailed operation references:
 
@@ -136,6 +137,7 @@ Detailed operation references:
 - [VSpin centrifuge and Access2 loader](../agilent/vspin/events.md)
 - [Diagnostic transports](event-bus/diagnostic-transports.md)
 - [Canonical schema for every operation above](../../contributor_guide/event-schemas.md)
+- [Manual operator actions](manual-operator.md#eventbus-integration)
 
 ```{toctree}
 :hidden:
diff --git a/docs/user_guide/machine-agnostic-features/manual-operator.md b/docs/user_guide/machine-agnostic-features/manual-operator.md
new file mode 100644
index 00000000000..005895fdb52
--- /dev/null
+++ b/docs/user_guide/machine-agnostic-features/manual-operator.md
@@ -0,0 +1,131 @@
+# Operator actions
+
+`ManualOperator` lets a protocol await work performed by a person without coupling the protocol
+to a terminal, notebook, graphical interface, or external task system.
+
+## Why use a manual operator?
+
+Many protocols start with a direct `input()` call. That is a reasonable choice for a simple
+notebook pause, but `ManualOperator` gives the same physical handoff a reusable PLR contract:
+
+- **Provider independence:** keep the protocol code unchanged while presenting the request through
+  terminal input, a notebook, a custom GUI, LIMS, Slack, or another acknowledgement system.
+- **Explicit outcomes:** providers report `completed`, `cancelled`, or `failed` rather than
+  reducing every acknowledgement to Enter being pressed.
+- **Resource-model reconciliation:** `move_resource()` validates a manual transfer before and
+  after acknowledgement, then applies the corresponding PLR assignment only when the model is
+  still consistent.
+- **Native traceability:** an active EventBus receives a correlated lifecycle with the action,
+  affected resources, endpoints, operator acknowledgement, and failure details where applicable.
+
+The built-in console provider is intentionally simple. It is a practical default for users who
+only need an interactive pause; more capable providers are optional application integrations.
+
+```python
+from pylabrobot.manual_operator import ConsoleOperatorActionProvider, ManualOperator
+
+operator = ManualOperator(ConsoleOperatorActionProvider(), name="cell_operator")
+
+await operator.perform(
+  action="centrifuge.spin",
+  title="Spin sample plate",
+  instructions="Spin plate_1 at 300 x g for 180 seconds, then return it to the handover nest.",
+  details={
+    "relative_centrifugal_force": 300,
+    "duration": 180,
+  },
+)
+```
+
+The built-in console provider treats Enter as a successful acknowledgement. Applications can
+implement `OperatorActionProvider` to present the same request through a graphical interface,
+HTTP service, LIMS, or message broker:
+
+```python
+from pylabrobot.manual_operator import OperatorActionRequest, OperatorActionResult
+
+
+class DashboardOperatorActionProvider:
+  async def request(self, action: OperatorActionRequest) -> OperatorActionResult:
+    result = await dashboard.publish_and_wait(action)
+    return OperatorActionResult.completed(confirmed_by=result.user)
+```
+
+`OperatorActionRequest` carries any direct modeled `resources` plus optional `source` and
+`destination` endpoints. Providers that cross a process boundary can serialize those objects in
+their own transport format; `details` remains for operation-specific request data.
+
+For a complete notebook example using a chatterbox incubator, manual plate transfer, plate reader,
+and optional EventBus subscriber, see the
+[ManualOperator Jupyter cookbook](../../cookbook/manual_operator_jupyter.ipynb).
+
+Providers return one of three explicit outcomes:
+
+- `completed`: `perform()` returns the result and the protocol continues.
+- `cancelled`: `perform()` raises `OperatorActionCancelledError`.
+- `failed`: `perform()` raises `OperatorActionFailedError`.
+
+An acknowledgement records what the operator reported. It does not prove that the physical work
+occurred. Protocol-specific validation and resource-model reconciliation should happen before or
+after `perform()` as appropriate.
+
+## Moving a resource
+
+Use `move_resource()` when the manual action transfers a modeled PLR resource between two modeled
+locations:
+
+```python
+from pylabrobot.resources import Rotation
+
+await operator.move_resource(
+  resource=sample_plate,
+  source=centrifuge_loader,
+  destination=handover_nest,
+  destination_rotation=Rotation(z=0),
+)
+```
+
+The method validates the source and destination before prompting, but leaves the resource assigned
+to its source while the operator works. After the provider reports completion, it validates the
+model again, detaches the resource from its source, and assigns it to the destination using PLR's
+normal resource-assignment machinery. This supports transfers between holders on the same deck
+root. A `ResourceHolder` supplies its normal child location; other destinations can use an
+explicit `destination_location=Coordinate(...)`. Pass an explicit
+`destination_rotation=Rotation(...)` when the manual move changes orientation. The rotation is
+never inferred from the destination holder and is applied before a holder calculates its child
+location. If final assignment fails, the original parent, location, and rotation are restored.
+
+`destination_rotation` is the resource's local rotation relative to `destination`, not an
+absolute/world rotation. After assignment, the resource's absolute rotation is composed from the
+destination's absolute rotation and this local rotation. For a manual move that replaces an
+automated transfer, pass the destination-local pose that the automated transfer would produce; do
+not copy the source resource's local rotation unless that is physically correct for the
+destination.
+
+Cancellation, reported failure, and provider exceptions do not modify the resource model. If the
+model changes while the operator request is pending, the method raises an error rather than
+overwriting the newer state.
+
+## EventBus integration
+
+When an EventBus subscriber is active, each awaited action emits one correlated lifecycle using
+the requested action as its semantic subtype:
+
+```text
+manual_operator.centrifuge.spin.started
+manual_operator.centrifuge.spin.completed
+```
+
+The event identifies the `ManualOperator` as `device`, includes any direct PLR `resources` passed
+to `perform()`, and preserves the request's title, instructions, confirmation text, and structured
+details. The completed event adds `confirmed_by` and the provider's result message when supplied.
+Cancellation, reported failure, invalid provider results, and provider exceptions emit `.failed`
+with the normal EventBus error fields.
+
+Use stable action identifiers such as `centrifuge.spin`, `plate_reader.read`, or
+`quality_control.inspect`. When the manual action has an automated counterpart, use that
+operation's canonical field names and PLR default units inside `details`. `move_resource()` emits
+`manual_operator.resource.move.*` with the direct moved resource plus its true `source` and
+`destination` resource references; do not mirror those endpoints as free-form `details`. The normal
+`resource.unassigned` and `resource.assigned` state-transition events record the subsequent model
+update.
diff --git a/pylabrobot/manual_operator/__init__.py b/pylabrobot/manual_operator/__init__.py
new file mode 100644
index 00000000000..6cf17daf397
--- /dev/null
+++ b/pylabrobot/manual_operator/__init__.py
@@ -0,0 +1,24 @@
+"""Awaitable operator actions for manual protocol steps."""
+
+from .operator import ManualOperator
+from .provider import ConsoleOperatorActionProvider, OperatorActionProvider
+from .standard import (
+  OperatorActionCancelledError,
+  OperatorActionError,
+  OperatorActionFailedError,
+  OperatorActionRequest,
+  OperatorActionResult,
+  OperatorActionStatus,
+)
+
+__all__ = [
+  "ConsoleOperatorActionProvider",
+  "ManualOperator",
+  "OperatorActionCancelledError",
+  "OperatorActionError",
+  "OperatorActionFailedError",
+  "OperatorActionProvider",
+  "OperatorActionRequest",
+  "OperatorActionResult",
+  "OperatorActionStatus",
+]
diff --git a/pylabrobot/manual_operator/manual_operator_tests.py b/pylabrobot/manual_operator/manual_operator_tests.py
new file mode 100644
index 00000000000..75d2f5f7042
--- /dev/null
+++ b/pylabrobot/manual_operator/manual_operator_tests.py
@@ -0,0 +1,458 @@
+import asyncio
+import unittest
+from typing import Callable
+
+from pylabrobot.events import EventBus, PLREvent, use_event_bus
+from pylabrobot.manual_operator import (
+  ConsoleOperatorActionProvider,
+  ManualOperator,
+  OperatorActionCancelledError,
+  OperatorActionFailedError,
+  OperatorActionRequest,
+  OperatorActionResult,
+)
+from pylabrobot.resources import Coordinate, Resource, ResourceHolder, Rotation
+
+
+class RecordingProvider:
+  def __init__(self, result: OperatorActionResult):
+    self.result = result
+    self.requests: list[OperatorActionRequest] = []
+
+  async def request(self, action: OperatorActionRequest) -> OperatorActionResult:
+    self.requests.append(action)
+    return self.result
+
+
+class CallbackProvider:
+  def __init__(self, callback: Callable[[OperatorActionRequest], OperatorActionResult]):
+    self.callback = callback
+    self.requests: list[OperatorActionRequest] = []
+
+  async def request(self, action: OperatorActionRequest) -> OperatorActionResult:
+    self.requests.append(action)
+    return self.callback(action)
+
+
+class TestManualOperator(unittest.IsolatedAsyncioTestCase):
+  async def test_perform_sends_structured_request_and_returns_completion(self):
+    provider = RecordingProvider(OperatorActionResult.completed(confirmed_by="operator-1"))
+    operator = ManualOperator(provider, name="cell_operator")
+
+    result = await operator.perform(
+      action="centrifuge.spin",
+      title="Spin sample plate",
+      instructions="Spin plate_1 at 300 x g for 180 seconds.",
+      confirmation_text="Confirm spin completed",
+      details={"relative_centrifugal_force": 300, "duration": 180},
+    )
+
+    self.assertEqual(result.confirmed_by, "operator-1")
+    self.assertEqual(len(provider.requests), 1)
+    request = provider.requests[0]
+    self.assertEqual(request.operator_name, "cell_operator")
+    self.assertEqual(request.action, "centrifuge.spin")
+    self.assertEqual(request.details["duration"], 180)
+    self.assertEqual(request.resources, ())
+    self.assertIsNone(request.source)
+    self.assertIsNone(request.destination)
+
+  async def test_cancelled_result_raises_specific_error(self):
+    provider = RecordingProvider(OperatorActionResult.cancelled(message="Protocol stopped"))
+    operator = ManualOperator(provider)
+
+    with self.assertRaisesRegex(OperatorActionCancelledError, "Protocol stopped"):
+      await operator.perform(action="inspect", title="Inspect plate", instructions="Inspect it.")
+
+  async def test_failed_result_raises_specific_error(self):
+    provider = RecordingProvider(OperatorActionResult.failed(message="Plate was damaged"))
+    operator = ManualOperator(provider)
+
+    with self.assertRaisesRegex(OperatorActionFailedError, "Plate was damaged"):
+      await operator.perform(action="inspect", title="Inspect plate", instructions="Inspect it.")
+
+  async def test_provider_exception_propagates(self):
+    class FailingProvider:
+      async def request(self, action: OperatorActionRequest) -> OperatorActionResult:
+        del action
+        raise ConnectionError("provider disconnected")
+
+    with self.assertRaisesRegex(ConnectionError, "provider disconnected"):
+      await ManualOperator(FailingProvider()).perform(
+        action="inspect", title="Inspect plate", instructions="Inspect it."
+      )
+
+  async def test_request_copies_details(self):
+    details = {"duration": 60}
+    resources = [Resource("plate", size_x=80, size_y=60, size_z=15)]
+    request = OperatorActionRequest(
+      operator_name="operator",
+      action="centrifuge.spin",
+      title="Spin",
+      instructions="Spin the plate.",
+      details=details,
+      resources=resources,
+    )
+
+    details["duration"] = 120
+    resources.clear()
+
+    self.assertEqual(request.details["duration"], 60)
+    self.assertEqual(len(request.resources), 1)
+
+  async def test_result_rejects_unknown_status(self):
+    with self.assertRaisesRegex(ValueError, "Unsupported operator action status"):
+      OperatorActionResult(status="unknown")  # type: ignore[arg-type]
+
+  async def test_move_resource_reassigns_only_after_completion(self):
+    source = ResourceHolder("source", size_x=100, size_y=100, size_z=10)
+    destination = ResourceHolder(
+      "destination",
+      size_x=100,
+      size_y=100,
+      size_z=10,
+      child_location=Coordinate(4, 5, 6),
+    )
+    plate = Resource("plate", size_x=80, size_y=60, size_z=15)
+    source.assign_child_resource(plate)
+
+    def complete_while_model_is_unchanged(
+      request: OperatorActionRequest,
+    ) -> OperatorActionResult:
+      self.assertIs(plate.parent, source)
+      self.assertIs(source.resource, plate)
+      self.assertIsNone(destination.resource)
+      self.assertEqual(request.action, "resource.move")
+      return OperatorActionResult.completed(confirmed_by="operator-1")
+
+    provider = CallbackProvider(complete_while_model_is_unchanged)
+    result = await ManualOperator(provider).move_resource(
+      resource=plate,
+      source=source,
+      destination=destination,
+    )
+
+    self.assertEqual(result.confirmed_by, "operator-1")
+    self.assertIsNone(source.resource)
+    self.assertIs(destination.resource, plate)
+    self.assertEqual(plate.location, Coordinate(4, 5, 6))
+    request = provider.requests[0]
+    self.assertEqual(request.details, {})
+    self.assertEqual(request.resources, (plate,))
+    self.assertIs(request.source, source)
+    self.assertIs(request.destination, destination)
+
+  async def test_move_resource_reassigns_between_holders_on_the_same_root(self):
+    deck = Resource("deck", size_x=1000, size_y=1000, size_z=10)
+    source = ResourceHolder("plate_module_0", size_x=100, size_y=100, size_z=10)
+    destination = ResourceHolder("handover_nest", size_x=100, size_y=100, size_z=10)
+    deck.assign_child_resource(source, location=Coordinate(0, 0, 0))
+    deck.assign_child_resource(destination, location=Coordinate(200, 0, 0))
+    plate = Resource("plate", size_x=80, size_y=60, size_z=15)
+    source.assign_child_resource(plate)
+
+    await ManualOperator(RecordingProvider(OperatorActionResult.completed())).move_resource(
+      resource=plate,
+      source=source,
+      destination=destination,
+    )
+
+    self.assertIsNone(source.resource)
+    self.assertIs(destination.resource, plate)
+    self.assertIs(plate.parent, destination)
+
+  async def test_move_resource_uses_explicit_destination_location(self):
+    source = ResourceHolder("source", size_x=100, size_y=100, size_z=10)
+    destination = Resource("destination", size_x=100, size_y=100, size_z=10)
+    plate = Resource("plate", size_x=80, size_y=60, size_z=15)
+    source.assign_child_resource(plate)
+    location = Coordinate(1, 2, 3)
+    provider = RecordingProvider(OperatorActionResult.completed())
+
+    await ManualOperator(provider).move_resource(
+      resource=plate,
+      source=source,
+      destination=destination,
+      destination_location=location,
+      details={"reason": "manual handoff"},
+    )
+
+    self.assertIs(plate.parent, destination)
+    self.assertEqual(plate.location, location)
+    self.assertEqual(provider.requests[0].details["reason"], "manual handoff")
+    self.assertEqual(
+      provider.requests[0].details["destination_location"],
+      {"x": 1, "y": 2, "z": 3, "type": "Coordinate"},
+    )
+
+  async def test_move_resource_applies_destination_rotation_before_assignment(self):
+    source = ResourceHolder("source", size_x=100, size_y=100, size_z=10)
+    destination = ResourceHolder(
+      "destination",
+      size_x=100,
+      size_y=100,
+      size_z=10,
+      child_location=Coordinate(4, 5, 6),
+    )
+    plate = Resource("plate", size_x=127, size_y=85, size_z=15, rotation=Rotation(z=90))
+    source.assign_child_resource(plate)
+    provider = RecordingProvider(OperatorActionResult.completed())
+
+    await ManualOperator(provider).move_resource(
+      resource=plate,
+      source=source,
+      destination=destination,
+      destination_rotation=Rotation(z=0),
+    )
+
+    self.assertIs(plate.parent, destination)
+    self.assertEqual(plate.rotation.z, 0)
+    self.assertEqual(plate.location, Coordinate(4, 5, 6))
+    self.assertEqual(
+      provider.requests[0].details["destination_rotation"],
+      {"x": 0, "y": 0, "z": 0, "type": "Rotation"},
+    )
+
+  async def test_move_resource_cancellation_leaves_model_unchanged(self):
+    source = ResourceHolder("source", size_x=100, size_y=100, size_z=10)
+    destination = ResourceHolder("destination", size_x=100, size_y=100, size_z=10)
+    plate = Resource("plate", size_x=80, size_y=60, size_z=15, rotation=Rotation(z=90))
+    source.assign_child_resource(plate)
+    provider = RecordingProvider(OperatorActionResult.cancelled())
+
+    with self.assertRaises(OperatorActionCancelledError):
+      await ManualOperator(provider).move_resource(
+        resource=plate,
+        source=source,
+        destination=destination,
+        destination_rotation=Rotation(z=0),
+      )
+
+    self.assertIs(source.resource, plate)
+    self.assertIsNone(destination.resource)
+    self.assertEqual(plate.rotation.z, 90)
+
+  async def test_move_resource_restores_original_pose_when_destination_assignment_fails(self):
+    deck = Resource("deck", size_x=1000, size_y=1000, size_z=10)
+    source = ResourceHolder("source", size_x=100, size_y=100, size_z=10)
+    destination = ResourceHolder("destination", size_x=100, size_y=100, size_z=10)
+    deck.assign_child_resource(source, location=Coordinate(0, 0, 0))
+    deck.assign_child_resource(destination, location=Coordinate(200, 0, 0))
+    plate = Resource("plate", size_x=127, size_y=85, size_z=15, rotation=Rotation(z=90))
+    source.assign_child_resource(plate)
+    original_location = plate.location
+
+    def fail_assignment(_resource: Resource) -> None:
+      raise RuntimeError("destination assignment failed")
+
+    destination.register_will_assign_resource_callback(fail_assignment)
+
+    with self.assertRaisesRegex(RuntimeError, "destination assignment failed"):
+      await ManualOperator(RecordingProvider(OperatorActionResult.completed())).move_resource(
+        resource=plate,
+        source=source,
+        destination=destination,
+        destination_rotation=Rotation(z=0),
+      )
+
+    self.assertIs(source.resource, plate)
+    self.assertIs(plate.parent, source)
+    self.assertEqual(plate.location, original_location)
+    self.assertEqual(plate.rotation.z, 90)
+    self.assertIsNone(destination.resource)
+
+  async def test_move_resource_rejects_incorrect_source_before_prompt(self):
+    actual_source = ResourceHolder("actual_source", size_x=100, size_y=100, size_z=10)
+    stated_source = ResourceHolder("stated_source", size_x=100, size_y=100, size_z=10)
+    destination = ResourceHolder("destination", size_x=100, size_y=100, size_z=10)
+    plate = Resource("plate", size_x=80, size_y=60, size_z=15)
+    actual_source.assign_child_resource(plate)
+    provider = RecordingProvider(OperatorActionResult.completed())
+
+    with self.assertRaisesRegex(ValueError, "not source 'stated_source'"):
+      await ManualOperator(provider).move_resource(
+        resource=plate,
+        source=stated_source,
+        destination=destination,
+      )
+
+    self.assertEqual(provider.requests, [])
+    self.assertIs(plate.parent, actual_source)
+
+  async def test_move_resource_rejects_occupied_destination_before_prompt(self):
+    source = ResourceHolder("source", size_x=100, size_y=100, size_z=10)
+    destination = ResourceHolder("destination", size_x=100, size_y=100, size_z=10)
+    plate = Resource("plate", size_x=80, size_y=60, size_z=15)
+    other_plate = Resource("other_plate", size_x=80, size_y=60, size_z=15)
+    source.assign_child_resource(plate)
+    destination.assign_child_resource(other_plate)
+    provider = RecordingProvider(OperatorActionResult.completed())
+
+    with self.assertRaisesRegex(RuntimeError, "already has a resource"):
+      await ManualOperator(provider).move_resource(
+        resource=plate,
+        source=source,
+        destination=destination,
+      )
+
+    self.assertEqual(provider.requests, [])
+    self.assertIs(source.resource, plate)
+    self.assertIs(destination.resource, other_plate)
+
+  async def test_move_resource_detects_model_change_while_pending(self):
+    source = ResourceHolder("source", size_x=100, size_y=100, size_z=10)
+    destination = ResourceHolder("destination", size_x=100, size_y=100, size_z=10)
+    unexpected = ResourceHolder("unexpected", size_x=100, size_y=100, size_z=10)
+    plate = Resource("plate", size_x=80, size_y=60, size_z=15)
+    source.assign_child_resource(plate)
+
+    def change_model_then_complete(_request: OperatorActionRequest) -> OperatorActionResult:
+      unexpected.assign_child_resource(plate)
+      return OperatorActionResult.completed()
+
+    provider = CallbackProvider(change_model_then_complete)
+    with self.assertRaisesRegex(RuntimeError, "model changed while the manual move was pending"):
+      await ManualOperator(provider).move_resource(
+        resource=plate,
+        source=source,
+        destination=destination,
+      )
+
+    self.assertIs(plate.parent, unexpected)
+    self.assertIsNone(destination.resource)
+
+  async def test_perform_emits_action_specific_lifecycle_events(self):
+    plate = Resource("plate", size_x=80, size_y=60, size_z=15)
+    provider = RecordingProvider(
+      OperatorActionResult.completed(message="Spin verified", confirmed_by="operator-1")
+    )
+    event_bus = EventBus()
+    events: list[PLREvent] = []
+    event_bus.subscribe(events.append)
+
+    with use_event_bus(event_bus):
+      await ManualOperator(provider, name="cell_operator").perform(
+        action="centrifuge.spin",
+        title="Spin sample plate",
+        instructions="Spin plate at 300 x g for 180 seconds.",
+        details={"relative_centrifugal_force": 300, "duration": 180},
+        resources=[plate],
+      )
+
+    self.assertEqual(
+      [event.name for event in events],
+      [
+        "manual_operator.centrifuge.spin.started",
+        "manual_operator.centrifuge.spin.completed",
+      ],
+    )
+    self.assertEqual(events[0].context["operation"], "manual_operator.centrifuge.spin")
+    self.assertEqual(events[0].context["operation_id"], events[1].context["operation_id"])
+    self.assertEqual(
+      events[0].data["device"],
+      {"name": "cell_operator", "type": "ManualOperator"},
+    )
+    self.assertEqual(events[0].data["resources"][0]["name"], "plate")
+    self.assertEqual(events[0].data["manual_action"], "centrifuge.spin")
+    self.assertEqual(events[0].data["details"]["relative_centrifugal_force"], 300)
+    self.assertNotIn("confirmed_by", events[0].data)
+    self.assertEqual(events[1].data["confirmed_by"], "operator-1")
+    self.assertEqual(events[1].data["result_message"], "Spin verified")
+
+  async def test_perform_emits_failed_event_for_reported_failure(self):
+    provider = RecordingProvider(OperatorActionResult.failed(message="Inspection failed"))
+    event_bus = EventBus()
+    events: list[PLREvent] = []
+    event_bus.subscribe(events.append)
+
+    with use_event_bus(event_bus):
+      with self.assertRaises(OperatorActionFailedError):
+        await ManualOperator(provider).perform(
+          action="quality_control.inspect",
+          title="Inspect sample",
+          instructions="Inspect the sample.",
+        )
+
+    self.assertEqual(
+      [event.name for event in events],
+      [
+        "manual_operator.quality_control.inspect.started",
+        "manual_operator.quality_control.inspect.failed",
+      ],
+    )
+    self.assertEqual(events[1].data["error_type"], "OperatorActionFailedError")
+    self.assertEqual(events[1].data["error_message"], "Inspection failed")
+
+  async def test_move_resource_emits_resource_and_endpoint_context(self):
+    source = ResourceHolder("source", size_x=100, size_y=100, size_z=10)
+    destination = ResourceHolder("destination", size_x=100, size_y=100, size_z=10)
+    plate = Resource("plate", size_x=80, size_y=60, size_z=15)
+    source.assign_child_resource(plate)
+    provider = RecordingProvider(OperatorActionResult.completed())
+    event_bus = EventBus()
+    events: list[PLREvent] = []
+    event_bus.subscribe(events.append)
+
+    with use_event_bus(event_bus):
+      await ManualOperator(provider).move_resource(
+        resource=plate,
+        source=source,
+        destination=destination,
+        destination_rotation=Rotation(z=90),
+      )
+
+    manual_events = [event for event in events if event.name.startswith("manual_operator.")]
+    self.assertEqual(
+      [event.name for event in manual_events],
+      ["manual_operator.resource.move.started", "manual_operator.resource.move.completed"],
+    )
+    self.assertEqual(manual_events[0].data["resources"][0]["name"], "plate")
+    self.assertEqual(manual_events[0].data["source"]["name"], "source")
+    self.assertEqual(manual_events[0].data["destination"]["name"], "destination")
+    self.assertEqual(
+      manual_events[0].data["details"]["destination_rotation"],
+      {"x": 0, "y": 0, "z": 90, "type": "Rotation"},
+    )
+    self.assertEqual(
+      manual_events[1].data["details"]["destination_rotation"],
+      {"x": 0, "y": 0, "z": 90, "type": "Rotation"},
+    )
+    self.assertEqual(
+      [event.name for event in events if event.name.startswith("resource.")],
+      ["resource.unassigned", "resource.assigned"],
+    )
+    self.assertIs(plate.parent, destination)
+
+
+class TestConsoleOperatorActionProvider(unittest.IsolatedAsyncioTestCase):
+  async def test_enter_completes_action_without_blocking_event_loop(self):
+    output: list[str] = []
+
+    def input_fn(prompt: str) -> str:
+      output.append(prompt)
+      return ""
+
+    provider = ConsoleOperatorActionProvider(
+      input_fn=input_fn,
+      output_fn=output.append,
+    )
+    request = OperatorActionRequest(
+      operator_name="operator",
+      action="inspect",
+      title="Inspect plate",
+      instructions="Check that the plate is seated.",
+    )
+    event_loop_progressed = asyncio.Event()
+
+    async def mark_progress() -> None:
+      await asyncio.sleep(0)
+      event_loop_progressed.set()
+
+    progress_task = asyncio.create_task(mark_progress())
+    result = await provider.request(request)
+    await progress_task
+
+    self.assertTrue(event_loop_progressed.is_set())
+    self.assertEqual(result, OperatorActionResult.completed())
+    self.assertIn("Inspect plate", output[0])
+    self.assertEqual(output[1], "Confirm action completed: ")
diff --git a/pylabrobot/manual_operator/operator.py b/pylabrobot/manual_operator/operator.py
new file mode 100644
index 00000000000..23c6d181bcb
--- /dev/null
+++ b/pylabrobot/manual_operator/operator.py
@@ -0,0 +1,221 @@
+"""Protocol-facing frontend for awaiting manual operator actions."""
+
+from typing import Any, Dict, Optional, Sequence
+
+from pylabrobot.events import device_reference, event_operation, resource_reference
+from pylabrobot.resources import Coordinate, Resource, Rotation
+
+from .provider import OperatorActionProvider
+from .standard import (
+  OperatorActionCancelledError,
+  OperatorActionFailedError,
+  OperatorActionRequest,
+  OperatorActionResult,
+)
+
+
+class ManualOperator:
+  """Await manual protocol work through a pluggable acknowledgement provider."""
+
+  def __init__(self, provider: OperatorActionProvider, name: str = "operator"):
+    if not name.strip():
+      raise ValueError("name must not be empty")
+    self.name = name
+    self.provider = provider
+
+  async def perform(
+    self,
+    *,
+    action: str,
+    title: str,
+    instructions: str,
+    confirmation_text: str = "Confirm action completed",
+    details: Optional[Dict[str, Any]] = None,
+    resources: Optional[Sequence[Resource]] = None,
+    source: Optional[Resource] = None,
+    destination: Optional[Resource] = None,
+  ) -> OperatorActionResult:
+    """Await one operator action and return its successful acknowledgement.
+
+    Providers report a structured outcome. Cancellation and failure become distinct exceptions;
+    exceptions raised by the provider itself propagate unchanged.
+    """
+
+    request = OperatorActionRequest(
+      operator_name=self.name,
+      action=action,
+      title=title,
+      instructions=instructions,
+      confirmation_text=confirmation_text,
+      details={} if details is None else details,
+      resources=resources or (),
+      source=source,
+      destination=destination,
+    )
+    operation_data: Dict[str, Any] = {
+      "device": device_reference(self, name=self.name),
+      "resources": [resource_reference(resource) for resource in request.resources],
+      "manual_action": request.action,
+      "title": request.title,
+      "instructions": request.instructions,
+      "confirmation_text": request.confirmation_text,
+      "details": request.details.copy(),
+    }
+    if source is not None:
+      operation_data["source"] = resource_reference(source)
+    if destination is not None:
+      operation_data["destination"] = resource_reference(destination)
+
+    result: Optional[OperatorActionResult] = None
+
+    def completed_data() -> Dict[str, Any]:
+      assert result is not None
+      data = operation_data.copy()
+      if result.message is not None:
+        data["result_message"] = result.message
+      if result.confirmed_by is not None:
+        data["confirmed_by"] = result.confirmed_by
+      return data
+
+    with event_operation(
+      f"manual_operator.{request.action}",
+      completed_data_factory=completed_data,
+      **operation_data,
+    ):
+      result = await self.provider.request(request)
+
+      if not isinstance(result, OperatorActionResult):
+        raise TypeError("OperatorActionProvider.request() must return OperatorActionResult")
+      if result.status == "cancelled":
+        raise OperatorActionCancelledError(request, result)
+      if result.status == "failed":
+        raise OperatorActionFailedError(request, result)
+      if result.status != "completed":
+        raise ValueError(f"Unsupported operator action status: {result.status!r}")
+    assert result is not None
+    return result
+
+  async def move_resource(
+    self,
+    *,
+    resource: Resource,
+    source: Resource,
+    destination: Resource,
+    destination_location: Optional[Coordinate] = None,
+    destination_rotation: Optional[Rotation] = None,
+    title: Optional[str] = None,
+    instructions: Optional[str] = None,
+    confirmation_text: str = "Confirm resource moved",
+    details: Optional[Dict[str, Any]] = None,
+  ) -> OperatorActionResult:
+    """Await a manual resource move, then reconcile the PLR resource model.
+
+    The resource remains assigned to ``source`` while the operator action is pending. A completed
+    acknowledgement reassigns it to ``destination`` using PLR's normal assignment machinery.
+    Cancellation, reported failure, and provider exceptions leave the model unchanged.
+
+    Args:
+      resource: Resource physically moved by the operator.
+      source: Resource currently containing ``resource``.
+      destination: Resource that will contain ``resource`` after the move.
+      destination_location: Optional resource location relative to ``destination``. Resource
+        holders calculate their normal child location when this is omitted.
+      destination_rotation: Optional resource-local rotation to apply before destination
+        assignment. It is relative to ``destination``, not an absolute/world rotation. After
+        assignment, PLR composes the destination's absolute rotation with this local rotation.
+        For a manual replacement of an automated transfer, pass the destination-local pose the
+        automated transfer would produce; do not copy the source local rotation unless it is
+        physically correct for ``destination``. This is not inferred from ``destination``.
+      title: Provider-facing title. Defaults to ``"Move <resource name>"``.
+      instructions: Provider-facing instructions. Defaults to a concise source-to-destination
+        instruction.
+      confirmation_text: Provider-facing completion acknowledgement text.
+      details: Additional transport-safe details for the provider.
+    """
+
+    self._validate_resource_move(
+      resource=resource,
+      source=source,
+      destination=destination,
+    )
+
+    requested_rotation: Optional[Rotation] = None
+    if destination_rotation is not None:
+      requested_rotation = Rotation(
+        x=destination_rotation.x,
+        y=destination_rotation.y,
+        z=destination_rotation.z,
+      )
+
+    move_details = {} if details is None else details.copy()
+    if destination_location is not None:
+      move_details["destination_location"] = destination_location.serialize()
+    if requested_rotation is not None:
+      move_details["destination_rotation"] = requested_rotation.serialize()
+
+    result = await self.perform(
+      action="resource.move",
+      title=title or f"Move {resource.name}",
+      instructions=instructions
+      or f"Move {resource.name} from {source.name} to {destination.name}.",
+      confirmation_text=confirmation_text,
+      details=move_details,
+      resources=[resource],
+      source=source,
+      destination=destination,
+    )
+
+    try:
+      self._validate_resource_move(
+        resource=resource,
+        source=source,
+        destination=destination,
+      )
+    except (RuntimeError, ValueError) as error:
+      raise RuntimeError(
+        "The PLR resource model changed while the manual move was pending; "
+        "the completed physical move was not applied to the model."
+      ) from error
+
+    previous_location = resource.location
+    previous_rotation = Rotation(
+      x=resource.rotation.x,
+      y=resource.rotation.y,
+      z=resource.rotation.z,
+    )
+    source.unassign_child_resource(resource)
+    try:
+      if requested_rotation is not None:
+        resource.rotation = requested_rotation
+      destination.assign_child_resource(
+        resource=resource,
+        location=destination_location,
+        reassign=True,
+      )
+    except Exception:
+      if resource.parent is not None:
+        resource.parent.unassign_child_resource(resource)
+      resource.rotation = previous_rotation
+      source.assign_child_resource(
+        resource=resource,
+        location=previous_location,
+        reassign=True,
+      )
+      raise
+    return result
+
+  @staticmethod
+  def _validate_resource_move(
+    *,
+    resource: Resource,
+    source: Resource,
+    destination: Resource,
+  ) -> None:
+    if resource.parent is not source:
+      current_parent = None if resource.parent is None else resource.parent.name
+      raise ValueError(
+        f"Resource {resource.name!r} is assigned to {current_parent!r}, not source {source.name!r}."
+      )
+    if source is destination:
+      raise ValueError("source and destination must be different resources")
+    destination.check_can_drop_resource_here(resource, reassign=True)
diff --git a/pylabrobot/manual_operator/provider.py b/pylabrobot/manual_operator/provider.py
new file mode 100644
index 00000000000..e019c23fa12
--- /dev/null
+++ b/pylabrobot/manual_operator/provider.py
@@ -0,0 +1,36 @@
+"""Provider interfaces and built-in transports for operator acknowledgement."""
+
+import asyncio
+from typing import Callable, Protocol
+
+from .standard import OperatorActionRequest, OperatorActionResult
+
+
+class OperatorActionProvider(Protocol):
+  """Transport an action request to an operator and await their reported outcome."""
+
+  async def request(self, action: OperatorActionRequest) -> OperatorActionResult:
+    """Wait until an operator reports an outcome for ``action``."""
+
+
+class ConsoleOperatorActionProvider:
+  """A terminal provider where pressing Enter reports successful completion.
+
+  The input call runs in a worker thread so other tasks on the protocol event loop can continue.
+  Applications with their own prompt lifecycle should implement :class:`OperatorActionProvider`
+  instead.
+  """
+
+  def __init__(
+    self,
+    *,
+    input_fn: Callable[[str], str] = input,
+    output_fn: Callable[[str], None] = print,
+  ):
+    self._input = input_fn
+    self._output = output_fn
+
+  async def request(self, action: OperatorActionRequest) -> OperatorActionResult:
+    self._output(f"\n{action.title}\n\n{action.instructions}\n")
+    await asyncio.to_thread(self._input, f"{action.confirmation_text}: ")
+    return OperatorActionResult.completed()
diff --git a/pylabrobot/manual_operator/standard.py b/pylabrobot/manual_operator/standard.py
new file mode 100644
index 00000000000..e7cb39fc206
--- /dev/null
+++ b/pylabrobot/manual_operator/standard.py
@@ -0,0 +1,91 @@
+"""Shared request, result, and error types for operator actions."""
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, Literal, Optional, Sequence
+
+from pylabrobot.resources import Resource
+
+OperatorActionStatus = Literal["completed", "cancelled", "failed"]
+"""Outcome reported by an operator-action provider."""
+
+
+@dataclass
+class OperatorActionRequest:
+  """A transport-independent request for a person to perform one action.
+
+  ``action`` identifies the kind of work. A provider may add its own transport-specific
+  correlation identifier when publishing the request through a GUI, API, or message broker.
+  """
+
+  operator_name: str
+  action: str
+  title: str
+  instructions: str
+  confirmation_text: str = "Confirm action completed"
+  details: Dict[str, Any] = field(default_factory=dict)
+  resources: Sequence[Resource] = field(default_factory=tuple)
+  source: Optional[Resource] = None
+  destination: Optional[Resource] = None
+
+  def __post_init__(self) -> None:
+    if not self.operator_name.strip():
+      raise ValueError("operator_name must not be empty")
+    if not self.action.strip():
+      raise ValueError("action must not be empty")
+    if not self.title.strip():
+      raise ValueError("title must not be empty")
+    if not self.instructions.strip():
+      raise ValueError("instructions must not be empty")
+    if not self.confirmation_text.strip():
+      raise ValueError("confirmation_text must not be empty")
+    self.details = self.details.copy()
+    self.resources = tuple(self.resources)
+
+
+@dataclass(frozen=True)
+class OperatorActionResult:
+  """The outcome reported by an operator-action provider."""
+
+  status: OperatorActionStatus
+  message: Optional[str] = None
+  confirmed_by: Optional[str] = None
+
+  def __post_init__(self) -> None:
+    if self.status not in ("completed", "cancelled", "failed"):
+      raise ValueError(f"Unsupported operator action status: {self.status!r}")
+
+  @classmethod
+  def completed(
+    cls, *, message: Optional[str] = None, confirmed_by: Optional[str] = None
+  ) -> "OperatorActionResult":
+    return cls(status="completed", message=message, confirmed_by=confirmed_by)
+
+  @classmethod
+  def cancelled(
+    cls, *, message: Optional[str] = None, confirmed_by: Optional[str] = None
+  ) -> "OperatorActionResult":
+    return cls(status="cancelled", message=message, confirmed_by=confirmed_by)
+
+  @classmethod
+  def failed(
+    cls, *, message: Optional[str] = None, confirmed_by: Optional[str] = None
+  ) -> "OperatorActionResult":
+    return cls(status="failed", message=message, confirmed_by=confirmed_by)
+
+
+class OperatorActionError(RuntimeError):
+  """Base exception raised when a manual operator action does not complete."""
+
+  def __init__(self, request: OperatorActionRequest, result: OperatorActionResult):
+    self.request = request
+    self.result = result
+    message = result.message or f"Operator action {request.action!r} did not complete."
+    super().__init__(message)
+
+
+class OperatorActionCancelledError(OperatorActionError):
+  """Raised when an operator cancels a requested action."""
+
+
+class OperatorActionFailedError(OperatorActionError):
+  """Raised when an operator reports that a requested action could not be completed."""