Docs(docs[engines]): Teach every engine

why: Engine users need transport-specific examples that expose real output
and boundaries without treating mock results as server evidence.

what:
- Add one executable first-success example per concrete engine
- Map every engine to a focused live or offline tutorial
- Test tutorial ownership and transport boundaries

Author: tony

docs/experimental/engines.md

@@ -8,15 +8,15 @@ boundary around execution.
 
 ## Choose an engine
 
-| Engine                                                                                                                        | Dispatch model                                        | Use it for                                                  |
-| ----------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------- | ----------------------------------------------------------- |
-| {class}`~libtmux.experimental.engines.subprocess.SubprocessEngine` ([reference](engines/subprocess.md))                       | One tmux CLI process per request                      | Straightforward synchronous live-server work                |
-| {class}`~libtmux.experimental.engines.asyncio.AsyncSubprocessEngine` ([reference](engines/async-subprocess.md))               | One asyncio subprocess per request                    | Async applications that do not need a persistent connection |
-| {class}`~libtmux.experimental.engines.mock.MockEngine` ([reference](engines/mock.md))                                         | Stateless in-memory simulation                        | Offline rendering and result-conversion tests               |
-| {class}`~libtmux.experimental.engines.mock.AsyncMockEngine` ([reference](engines/async-mock.md))                              | Async stateless in-memory simulation                  | Offline tests of async callers                              |
-| {class}`~libtmux.experimental.engines.control_mode.ControlModeEngine` ([reference](engines/control-mode.md))                  | Subprocess bootstrap, then persistent `tmux -C`       | Pipelined command batches                                   |
-| {class}`~libtmux.experimental.engines.async_control_mode.AsyncControlModeEngine` ([reference](engines/async-control-mode.md)) | Async subprocess bootstrap, then supervised `tmux -C` | Async batches and control-mode notifications                |
-| {class}`~libtmux.experimental.engines.imsg.base.ImsgEngine` ([reference](engines/imsg.md))                                    | Native socket with required CLI fallbacks             | POSIX protocol experiments and narrow parity checks         |
+| Engine                                                                                                                        | Dispatch model                                        | Use it for                                                  | Worked example                       |
+| ----------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------- | ----------------------------------------------------------- | ------------------------------------ |
+| {class}`~libtmux.experimental.engines.subprocess.SubprocessEngine` ([reference](engines/subprocess.md))                       | One tmux CLI process per request                      | Straightforward synchronous live-server work                | {doc}`tutorials/live-operation`      |
+| {class}`~libtmux.experimental.engines.asyncio.AsyncSubprocessEngine` ([reference](engines/async-subprocess.md))               | One asyncio subprocess per request                    | Async applications that do not need a persistent connection | {doc}`tutorials/async-subprocess`    |
+| {class}`~libtmux.experimental.engines.mock.MockEngine` ([reference](engines/mock.md))                                         | Stateless in-memory simulation                        | Offline rendering and result-conversion tests               | {doc}`tutorials/offline-testing`     |
+| {class}`~libtmux.experimental.engines.mock.AsyncMockEngine` ([reference](engines/async-mock.md))                              | Async stateless in-memory simulation                  | Offline tests of async callers                              | {doc}`tutorials/offline-testing`     |
+| {class}`~libtmux.experimental.engines.control_mode.ControlModeEngine` ([reference](engines/control-mode.md))                  | Subprocess bootstrap, then persistent `tmux -C`       | Pipelined command batches                                   | {doc}`tutorials/control-mode`        |
+| {class}`~libtmux.experimental.engines.async_control_mode.AsyncControlModeEngine` ([reference](engines/async-control-mode.md)) | Async subprocess bootstrap, then supervised `tmux -C` | Async batches and control-mode notifications                | {doc}`tutorials/async-control-plans` |
+| {class}`~libtmux.experimental.engines.imsg.base.ImsgEngine` ([reference](engines/imsg.md))                                    | Native socket with required CLI fallbacks             | POSIX protocol experiments and narrow parity checks         | {doc}`tutorials/imsg-parity`         |
 
 The control-mode engines open a persistent client only when an existing
 session has `destroy-unattached` set to `off`. Until then, they use their
@@ -30,13 +30,10 @@ event loop.
 
 ```python
 >>> from libtmux.experimental.engines import available_engines, create_engine
->>> from libtmux.experimental.ops import HasSession, run
->>> from libtmux.experimental.ops._types import SessionId
 >>> available_engines()
 ('control_mode', 'imsg', 'mock', 'subprocess')
->>> engine = create_engine("mock")
->>> run(HasSession(target=SessionId("$0")), engine).status
-'complete'
+>>> type(create_engine("mock")).__name__
+'MockEngine'
 ```
 
 ## Engine boundary
@@ -56,16 +53,18 @@ raise at the engine boundary.
 
 ## Tutorials
 
-Start with {doc}`tutorials/live-operation`, then use the task guide that matches
-the boundary you need to understand:
+Each engine row links to its tested workflow. Start with
+{doc}`tutorials/live-operation`, or go directly to the transport you need:
 
-- {doc}`tutorials/results-and-failures`
 - {doc}`tutorials/async-control-plans`
 - {doc}`tutorials/async-subprocess`
 - {doc}`tutorials/control-mode`
 - {doc}`tutorials/offline-testing`
 - {doc}`tutorials/imsg-parity`
 
+{doc}`tutorials/results-and-failures` is the shared result guide rather than an
+engine-owned workflow.
+
 ## Shared API
 
 ```{eval-rst}

docs/experimental/engines/async-control-mode.md

@@ -32,28 +32,23 @@ the first batch uses
 `new-session` bootstrap an empty server without creating a private control
 session. A later batch can open the persistent connection.
 
+Within the context, pass typed operations to
+{func}`~libtmux.experimental.ops.arun` as you would with any asynchronous
+engine.
+
 ```python
 >>> import asyncio
->>> from libtmux.experimental.engines import AsyncControlModeEngine, CommandRequest
->>> assert session.session_id is not None
->>> assert window.window_id is not None
->>> requests = [
-...     CommandRequest.from_args(
-...         "display-message", "-p", "-t", session.session_id, "#{session_id}"
-...     ),
-...     CommandRequest.from_args(
-...         "display-message", "-p", "-t", window.window_id, "#{window_id}"
-...     ),
-... ]
->>> async def query_targets():
+>>> from libtmux.experimental.engines import AsyncControlModeEngine
+>>> from libtmux.experimental.ops import DisplayMessage, PaneId, arun
+>>> assert pane is not None and pane.pane_id is not None
+>>> async def read_pane_id():
 ...     async with AsyncControlModeEngine.for_server(server) as engine:
-...         return await engine.run_batch(requests)
->>> results = asyncio.run(query_targets())
->>> [result.returncode for result in results]
-[0, 0]
->>> results[0].stdout == (session.session_id,)
-True
->>> results[1].stdout == (window.window_id,)
+...         result = await arun(
+...             DisplayMessage(target=PaneId(pane.pane_id), message="#{pane_id}"),
+...             engine,
+...         )
+...         return result.raise_for_status().text
+>>> asyncio.run(read_pane_id()) == pane.pane_id
 True
 ```
 
@@ -96,8 +91,7 @@ a notification emitted before the first iteration will be delivered.
    :special-members: __aenter__, __aexit__
 ```
 
-## Related tutorials
+## Related tutorial
 
-See {doc}`../tutorials/control-mode` for the shared batching contract and the
-notification boundary. See {doc}`../tutorials/async-control-plans` to compose
-forward-referenced operations and fold them into control-mode dispatches.
+See {doc}`../tutorials/async-control-plans` to compose forward-referenced
+operations and fold them into control-mode dispatches.

docs/experimental/engines/async-mock.md

@@ -15,19 +15,22 @@ reconnection, notifications, or real tmux state.
 
 ## Construction and cleanup
 
-Pass optional canned `capture_lines`. It owns no task, process, or connection,
-so no cleanup is required.
+Pass optional canned `capture_lines`. Create operations receive fabricated IDs.
+The engine owns no task, process, or connection, so no cleanup is required.
+
+The `%1` below is fabricated. The engine neither inspects `@404` nor creates a
+pane.
 
 ```python
 >>> import asyncio
 >>> from libtmux.experimental.engines import AsyncMockEngine
->>> from libtmux.experimental.ops import CapturePane, arun
->>> from libtmux.experimental.ops._types import PaneId
->>> async def capture_offline():
-...     engine = AsyncMockEngine(capture_lines=("offline",))
-...     return await arun(CapturePane(target=PaneId("%1")), engine)
->>> asyncio.run(capture_offline()).lines
-('offline',)
+>>> from libtmux.experimental.ops import SplitWindow, WindowId, arun
+>>> operation = SplitWindow(target=WindowId("@404"))
+>>> async def create_offline():
+...     return await arun(operation, AsyncMockEngine())
+>>> fabricated = asyncio.run(create_offline()).raise_for_status()
+>>> fabricated.new_pane_id, fabricated.argv == operation.render()
+('%1', True)
 ```
 
 ## Lifecycle and failure boundary
@@ -48,5 +51,5 @@ caller supplies `version`.
 
 ## Related tutorial
 
-See {doc}`../tutorials/offline-testing` for parallel sync and async test
-patterns.
+See {doc}`../tutorials/offline-testing` for synchronous and asynchronous
+offline test patterns.

docs/experimental/engines/async-subprocess.md

@@ -24,8 +24,7 @@ reaps its child.
 ```python
 >>> import asyncio
 >>> from libtmux.experimental.engines import AsyncSubprocessEngine
->>> from libtmux.experimental.ops import DisplayMessage, arun
->>> from libtmux.experimental.ops._types import PaneId
+>>> from libtmux.experimental.ops import DisplayMessage, PaneId, arun
 >>> assert pane is not None and pane.pane_id is not None
 >>> async def read_pane_id():
 ...     engine = AsyncSubprocessEngine.for_server(server)

docs/experimental/engines/control-mode.md

@@ -30,25 +30,19 @@ If no safe session exists, the batch uses
 This lets `new-session` bootstrap an empty server without creating a private
 control session. A later batch can open the persistent connection.
 
+Within the context, pass typed operations to
+{func}`~libtmux.experimental.ops.run` as you would with any synchronous engine.
+
 ```python
->>> from libtmux.experimental.engines import CommandRequest, ControlModeEngine
->>> assert session.session_id is not None
->>> assert window.window_id is not None
->>> requests = [
-...     CommandRequest.from_args(
-...         "display-message", "-p", "-t", session.session_id, "#{session_id}"
-...     ),
-...     CommandRequest.from_args(
-...         "display-message", "-p", "-t", window.window_id, "#{window_id}"
-...     ),
-... ]
+>>> from libtmux.experimental.engines import ControlModeEngine
+>>> from libtmux.experimental.ops import DisplayMessage, PaneId, run
+>>> assert pane is not None and pane.pane_id is not None
 >>> with ControlModeEngine.for_server(server) as engine:
-...     results = engine.run_batch(requests)
->>> [result.returncode for result in results]
-[0, 0]
->>> results[0].stdout == (session.session_id,)
-True
->>> results[1].stdout == (window.window_id,)
+...     result = run(
+...         DisplayMessage(target=PaneId(pane.pane_id), message="#{pane_id}"),
+...         engine,
+...     ).raise_for_status()
+>>> result.text == pane.pane_id
 True
 ```
 
@@ -78,5 +72,5 @@ new policy.
 
 ## Related tutorial
 
-See {doc}`../tutorials/control-mode` for sync and async batching and lifecycle
-differences.
+See {doc}`../tutorials/control-mode` for ordered pipelining over one synchronous
+control client.

docs/experimental/engines/imsg.md

@@ -26,31 +26,19 @@ in every
 
 ```python
 >>> from libtmux.experimental.engines import CommandRequest, ImsgEngine
->>> from libtmux.experimental.engines import SubprocessEngine
+>>> assert server.socket_name is not None
 >>> assert session.session_id is not None
->>> if server.socket_name is not None:
-...     prefix = (f"-L{server.socket_name}",)
-... else:
-...     socket_path = server.cmd(
-...         "display-message", "-p", "#{socket_path}"
-...     ).stdout[0]
-...     prefix = (f"-S{socket_path}",)
 >>> request = CommandRequest.from_args(
-...     *prefix,
+...     f"-L{server.socket_name}",
 ...     "display-message",
 ...     "-p",
 ...     "-t",
 ...     session.session_id,
 ...     "#{session_id}",
 ... )
->>> native = ImsgEngine().run(request)
->>> classic = SubprocessEngine().run(request)
->>> (
-...     native.returncode,
-...     native.stdout == classic.stdout,
-...     native.stdout[0] == session.session_id,
-... )
-(0, True, True)
+>>> result = ImsgEngine().run(request)
+>>> result.returncode, result.stdout == (session.session_id,)
+(0, True)
 ```
 
 ## Lifecycle and failure boundary

docs/experimental/engines/mock.md

@@ -18,17 +18,19 @@ permissions, shell execution, transport failures, or tmux output parsing.
 Pass optional canned `capture_lines`. An instance keeps only monotonic ID
 counters and needs no cleanup.
 
+The result below is canned. The engine does not contact tmux or inspect whether
+the target pane exists.
+
 ```python
 >>> from libtmux.experimental.engines import MockEngine
->>> from libtmux.experimental.ops import CapturePane, SplitWindow, run
->>> from libtmux.experimental.ops._types import PaneId, WindowId
->>> engine = MockEngine(capture_lines=("hello", "world"))
->>> operation = SplitWindow(target=WindowId("@1"))
->>> created = run(operation, engine)
->>> created.new_pane_id, created.argv == operation.render()
-('%1', True)
->>> run(CapturePane(target=PaneId("%1")), engine).lines
-('hello', 'world')
+>>> from libtmux.experimental.ops import CapturePane, PaneId, run
+>>> engine = MockEngine(capture_lines=("canned, no tmux",))
+>>> canned = run(
+...     CapturePane(target=PaneId("%404")),
+...     engine,
+... ).raise_for_status()
+>>> canned.lines
+('canned, no tmux',)
 ```
 
 ## Lifecycle and failure boundary

docs/experimental/engines/subprocess.md

@@ -23,6 +23,7 @@ process and has no cleanup method.
 ```python
 >>> from libtmux.experimental.engines import SubprocessEngine
 >>> from libtmux.experimental.ops import ListSessions, run
+>>> assert session.session_id is not None
 >>> engine = SubprocessEngine.for_server(server)
 >>> result = run(ListSessions(), engine).raise_for_status()
 >>> any(item.session_id == session.session_id for item in result.sessions)

docs/experimental/tutorials/async-control-plans.md

@@ -20,6 +20,15 @@ back. The pane does not exist when Python records the option operations, so
 {class}`~libtmux.experimental.ops._types.SlotRef` that stands in for its future
 ID.
 
+The same deterministic story gives an agent tool three explicit phases:
+
+- Observe: preview the unresolved plan and explain its two dispatch steps
+  without contacting tmux.
+- Act: execute the marked fold through
+  {class}`~libtmux.experimental.engines.async_control_mode.AsyncControlModeEngine`.
+- Verify: inspect four typed statuses, the captured `worker:ready` value, and
+  the live pane lookup.
+
 The two tabs show the same work at the Python and tmux boundaries. `@WINDOW`
 stands for the live window ID. `%PANE` stands for the pane ID captured by
 `split-window`.
@@ -207,6 +216,6 @@ before it can render the next target.
 - Scope the engine with `async with` so its reader, supervisor, and control
   client close before the event loop exits.
 
-See {doc}`control-mode` for attachment, notification, and reconnection
-lifecycle details. See {doc}`results-and-failures` for typed command failures
-and skipped work.
+See {doc}`../engines/async-control-mode` for attachment, notification, and
+reconnection lifecycle details. See {doc}`results-and-failures` for typed
+command failures and skipped work.

docs/experimental/tutorials/async-subprocess.md

@@ -3,7 +3,8 @@
 {class}`~libtmux.experimental.engines.asyncio.AsyncSubprocessEngine` uses
 {func}`asyncio.create_subprocess_exec` for command process I/O.
 {func}`~libtmux.experimental.ops.arun` preserves the same operation and typed
-result contract as synchronous execution.
+result contract as synchronous execution. The workflow below returns two typed,
+independent reads from one live server.
 
 ## Dispatch independent reads concurrently
 
@@ -18,8 +19,7 @@ would deliberately await them in order.
 ```python
 >>> import asyncio
 >>> from libtmux.experimental.engines import AsyncSubprocessEngine
->>> from libtmux.experimental.ops import DisplayMessage, arun
->>> from libtmux.experimental.ops._types import PaneId, SessionId
+>>> from libtmux.experimental.ops import DisplayMessage, PaneId, SessionId, arun
 >>> assert session.session_id is not None
 >>> assert pane is not None and pane.pane_id is not None
 >>> engine = AsyncSubprocessEngine.for_server(server)
@@ -45,11 +45,16 @@ would deliberately await them in order.
 ...         ),
 ...     )
 ...     return (
-...         session_result.raise_for_status().text,
-...         pane_result.raise_for_status().text,
+...         session_result.raise_for_status(),
+...         pane_result.raise_for_status(),
 ...     )
->>> asyncio.run(read_ids()) == (session.session_id, pane.pane_id)
-True
+>>> session_result, pane_result = asyncio.run(read_ids())
+>>> [type(result).__name__ for result in (session_result, pane_result)]
+['DisplayMessageResult', 'DisplayMessageResult']
+>>> session_result.status, pane_result.status
+('complete', 'complete')
+>>> session_result.text == session.session_id, pane_result.text == pane.pane_id
+(True, True)
 ```
 
 ## Cancellation boundary