diff --git a/README.md b/README.md
index 0073a11..b588cde 100644
--- a/README.md
+++ b/README.md
@@ -28,7 +28,7 @@ Ansible has a great executor and a huge module ecosystem, but the existing Pytho
 pip install ansible-host
 ```
 
-Requires Python 3.10+ and `ansible-core>=2.16,<2.20`. Like Ansible itself, the library runs on POSIX systems (Linux, macOS, WSL) — it is not supported on native Windows.
+Requires Python 3.10+ and `ansible-core>=2.16,<2.22`. Like Ansible itself, the library runs on POSIX systems (Linux, macOS, WSL) — it is not supported on native Windows.
 
 ## Development
 
@@ -58,32 +58,117 @@ uv run pytest
 
 ## Quickstart
 
+### 30-second try — no SSH, no inventory
+
+`AnsibleLocalhost` runs modules in-process on the current machine via Ansible's `local` connection plugin. No inventory file, no SSH, no setup.
+
 ```python
-from ansible_host import AnsibleHost, AnsibleHosts
+from ansible_host import AnsibleLocalhost
+
+host = AnsibleLocalhost()
+result = host.ping()
+assert result["ping"] == "pong"
 
-# Single host
-host = AnsibleHost(inventory="inventory.yml", pattern="vlab-01")
-result = host.shell("uptime")
+result = host.command("uname -a")
 print(result["stdout"])
+```
+
+### Running against a real host
+
+Drop an inventory file alongside your script:
 
-# Group of hosts (returns dict keyed by hostname)
-hosts = AnsibleHosts(inventory="inventory.yml", pattern="vms_1")
-results = hosts.ping()
+```ini
+# inventory.ini
+[switches]
+sw-01 ansible_host=10.0.0.1 ansible_user=admin
+```
+
+```python
+from ansible_host import AnsibleHost
+
+host = AnsibleHost(inventory="inventory.ini", pattern="sw-01")
+result = host.shell("show version")
+print(result["stdout"])
+```
+
+### Multi-host fanout
+
+```python
+from ansible_host import AnsibleHosts
+
+hosts = AnsibleHosts(inventory="inventory.ini", pattern="switches")
+results = hosts.ping()                       # parallelism via forks=
 for hostname, r in results.items():
     print(hostname, r["ping"])
 
-# Any Ansible module is callable as a method via __getattr__
+# Container API: index, iterate, len
+print(len(hosts), hosts.hostnames)
+first = hosts[0]                             # -> AnsibleHost
+by_name = hosts["sw-01"]                     # -> AnsibleHost
+for h in hosts:
+    print(h.hostname)
+```
+
+### Dynamic dispatch and task directives
+
+Any Ansible module is callable as a method via `__getattr__` (`host.<module_name>(...)`):
+
+```python
 host.copy(src="/etc/hosts", dest="/tmp/hosts.bak")
-host.shell("ls /tmp", task_directives={"ignore_errors": True})
+host.command("rm /tmp/maybe-missing", task_directives={"ignore_errors": True})
+host.shell("echo $TOKEN", task_directives={"no_log": True})
+```
+
+Common `task_directives`: `ignore_errors`, `no_log`, `when`, `failed_when`, `changed_when`, `become`.
+
+### Batch mode — queue tasks, run them in a single play
 
-# Batch mode: queue tasks, run them in a single play
+```python
 with host:
     host.shell("uptime")
     host.shell("df -h")
     host.shell("free -m")
-results = host.results  # list of task results
+results = host.results
+```
+
+### Building a queue across functions
+
+`with host:` is lexically scoped. If you need to assemble a batch across multiple functions, use the explicit form — same machinery, no scope limit:
+
+```python
+host.load_module("ansible.builtin.command", args=["uptime"])
+host.load_module("ansible.builtin.command", args=["df -h"])
+results = host.run_loaded_modules()
+```
+
+### Result shapes
+
+|              | single task             | batch (`with` block)                |
+| ------------ | ----------------------- | ----------------------------------- |
+| single host  | `dict`                  | `list[dict]`                        |
+| multi host   | `{hostname: dict}`      | `{hostname: list[dict]}`            |
+
+### Failures
+
+A failing module raises `AnsibleModuleFailed`:
+
+```python
+from ansible_host import AnsibleModuleFailed
+
+try:
+    host.command("false")
+except AnsibleModuleFailed as e:
+    print("module failed:", e)
+
+# Or suppress and inspect the result:
+result = host.command("false", task_directives={"ignore_errors": True})
+assert result["failed"] is True
 ```
 
+### More examples
+
+See [`tests/test_local_integration.py`](tests/test_local_integration.py) for 30 runnable examples covering ping, command, shell, batch mode, dynamic dispatch, `no_log`, `forks`, multi-host fanout, per-host failure aggregation, and the container protocol.
+
 ## Compatibility
 
 This library uses Ansible's internal Python API (`TaskQueueManager`, `InventoryManager`, `VariableManager`, `Play`, `DataLoader`). Those APIs are not officially stable across `ansible-core` releases — expect occasional updates when `ansible-core` introduces breaking internal changes. The current support range is declared in `pyproject.toml` and in the matrix CI.
diff --git a/src/ansible_host/__init__.py b/src/ansible_host/__init__.py
index 2b9eb53..057d095 100644
--- a/src/ansible_host/__init__.py
+++ b/src/ansible_host/__init__.py
@@ -30,6 +30,7 @@
 from ansible.module_utils.common.collections import ImmutableDict
 from ansible.parsing.dataloader import DataLoader
 from ansible.playbook.play import Play
+from ansible.plugins.callback import CallbackBase
 from ansible.plugins.loader import init_plugin_loader, module_loader
 from ansible.utils.display import Display
 from ansible.vars.hostvars import HostVars
@@ -42,6 +43,103 @@
 logger = logging.getLogger("ansible_pyapi")
 
 
+class _JsonResultsCallback(CallbackBase):
+    """Internal result collector for AnsibleHostsBase._run.
+
+    Not registered via Ansible's plugin loader: it is instantiated directly
+    by _run and attached to the TaskQueueManager's _callback_plugins list.
+    The leading underscore marks this as package-private; downstream users
+    should rely on the structured results returned by run_module() / .results
+    rather than instantiating this class themselves.
+
+    Derived from the json_results callback in sonic-net/sonic-mgmt (Apache 2.0).
+    """
+
+    CALLBACK_VERSION = 2.0
+    # Deliberately NOT 'stdout' — Ansible's built-in `null` stdout callback
+    # handles terminal output (silently); this collector is purely a result sink.
+    CALLBACK_TYPE = "notification"
+    CALLBACK_NAME = "ansible_host._json_results"
+
+    TASK_FIELDS = (
+        "action",
+        "become",
+        "become_method",
+        "become_user",
+        "connection",
+        "ignore_errors",
+        "ignore_unreachable",
+        "register",
+        "retries",
+        "timeout",
+    )
+
+    def __init__(self):
+        super().__init__()
+        self._results: dict[str, list[dict]] = {}
+
+    def _get_module_name(self, result):
+        if ansible.__version__ >= "2.19.0":
+            return result.task
+        return result.task_name
+
+    def _get_task_fields(self, result):
+        return {
+            field: result._task_fields[field]
+            for field in self.TASK_FIELDS
+            if field in result._task_fields
+        }
+
+    def _log_res(self, hostname, module_name, res):
+        if display.verbosity == 0:
+            return
+        if display.verbosity == 1:
+            log_func = display.v
+            brief = json.dumps(
+                {"module_name": module_name, "reachable": res["reachable"], "failed": res["failed"]},
+                default=str,
+            )
+            msg = f"[{hostname}] => {brief}"
+        elif display.verbosity == 2:
+            log_func = display.vv
+            msg = f"[{hostname}] => {json.dumps(res, default=str)}"
+        elif display.verbosity == 3:
+            log_func = display.vvv
+            msg = f"[{hostname}] => {json.dumps(res, indent=4, default=str)}"
+        else:
+            log_func = display.vvvv
+            msg = f"[{hostname}] => {json.dumps(res, indent=4, default=str)}"
+        log_func(msg)
+
+    def _record(self, result, *, reachable, failed):
+        hostname = str(result._host.get_name())
+        module_name = self._get_module_name(result)
+        self._results.setdefault(hostname, [])
+        res = dict(hostname=hostname, reachable=reachable, failed=failed)
+        # deepcopy preserves non-serializable values (e.g. Task instances in
+        # _task_fields on ansible-core 2.19+); the prior json round-trip lost
+        # or crashed on those.
+        res.update(copy.deepcopy(result._result))
+        if "invocation" in res and isinstance(res["invocation"], dict):
+            res["invocation"]["module_name"] = module_name
+        res["_task_fields"] = self._get_task_fields(result)
+        self._log_res(hostname, module_name, res)
+        self._results[hostname].append(res)
+
+    def v2_runner_on_ok(self, result):
+        self._record(result, reachable=True, failed=False)
+
+    def v2_runner_on_failed(self, result, *args, **kwargs):
+        self._record(result, reachable=True, failed=True)
+
+    def v2_runner_on_unreachable(self, result):
+        self._record(result, reachable=False, failed=True)
+
+    @property
+    def results(self) -> dict[str, list[dict]]:
+        return self._results
+
+
 def _to_native_type(value: Any) -> Any:
     """Convert Ansible types (AnsibleUnicode, AnsibleUnsafeText, etc.) to native Python types.
 
@@ -261,7 +359,8 @@ def _check_failed_results(self, results):
         if failed_results:
             raise AnsibleModuleFailed(
                 f"Ansible module failed. If failure is expected, use `task_directives={{'ignore_errors': True}}` "
-                f"to avoid raising an exception. Details: {json.dumps(failed_results, indent=4)}"
+                f"to avoid raising an exception. Details: "
+                f"{json.dumps(failed_results, indent=4, default=str)}"
             )
 
     def _run(
@@ -344,13 +443,18 @@ def _run(
                 variable_manager=self.vm,
                 loader=self.loader
             )
+            # TQM requires a registered stdout callback name to satisfy
+            # load_callbacks(); we pass 'minimal' (one of Ansible's built-ins)
+            # then evict it from _callback_plugins before run() so nothing
+            # prints to stdout. Our own _JsonResultsCallback is attached as
+            # the sole receiver of task events.
             if ansible.__version__ >= '2.19.0':
                 tqm = TaskQueueManager(
                     inventory=self.im,
                     variable_manager=self.vm,
                     loader=self.loader,
                     passwords={},
-                    stdout_callback_name='json_results',
+                    stdout_callback_name='minimal',
                     run_tree=False,
                     forks=self.options.get("forks")
                 )
@@ -360,16 +464,30 @@ def _run(
                     variable_manager=self.vm,
                     loader=self.loader,
                     passwords={},
-                    stdout_callback='json_results',
+                    stdout_callback='minimal',
                     run_tree=False,
                     forks=self.options.get("forks")
                 )
             tqm.load_callbacks()
 
+            # Remove the built-in stdout callback (suppresses terminal output)
+            # and attach our explicit result collector in its place.
+            tqm._callback_plugins[:] = [
+                cb for cb in tqm._callback_plugins
+                if getattr(cb, 'CALLBACK_TYPE', None) != 'stdout'
+            ]
+            results_collector = _JsonResultsCallback()
+            # _init_callback_methods populates _implemented_callback_methods,
+            # which TQM.send_callback gates dispatch on. The plugin loader calls
+            # this implicitly when loading callbacks via the loader; since we
+            # construct our collector directly, we have to invoke it ourselves.
+            if hasattr(results_collector, '_init_callback_methods'):
+                results_collector._init_callback_methods()
+            tqm._callback_plugins.append(results_collector)
+
             tqm.run(play)
 
-            stdout_callback = tqm._stdout_callback
-            results = stdout_callback.results
+            results = results_collector.results
 
             # results is a dict: {hostname: [task_result_dict, ...], ...}
             # It makes sense to return this format of results for multiple hosts and multiple tasks
diff --git a/tests/test_local_integration.py b/tests/test_local_integration.py
new file mode 100644
index 0000000..fbab208
--- /dev/null
+++ b/tests/test_local_integration.py
@@ -0,0 +1,415 @@
+"""Tier-1 integration tests using Ansible's `local` connection.
+
+These tests exercise the full execution path (build_task -> TaskQueueManager
+-> result parsing -> failure handling) without requiring any SSH listener or
+external host. They run the modules in-process on the test runner itself.
+
+If these tests fail, the library is broken for real users; if they pass, the
+SSH transport is the only piece left untested (covered by Tier-2 tests when
+they are added).
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from ansible_host import (
+    AnsibleHost,
+    AnsibleHosts,
+    AnsibleLocalhost,
+    AnsibleModuleFailed,
+    NoAnsibleHostError,
+)
+
+
+@pytest.fixture
+def host():
+    """A fresh AnsibleLocalhost per test.
+
+    AnsibleLocalhost defaults to connection: local, so no SSH is involved.
+    """
+    return AnsibleLocalhost()
+
+
+def test_ping_returns_pong(host):
+    result = host.run_module("ansible.builtin.ping")
+    assert isinstance(result, dict)
+    assert result.get("ping") == "pong"
+    assert result.get("failed") in (False, None)
+
+
+def test_command_module_returns_stdout(host):
+    result = host.run_module(
+        "ansible.builtin.command",
+        args=["echo hello-from-ansible-host"],
+    )
+    assert isinstance(result, dict)
+    assert result.get("rc") == 0
+    assert "hello-from-ansible-host" in result.get("stdout", "")
+
+
+def test_failing_command_raises(host):
+    with pytest.raises(AnsibleModuleFailed):
+        host.run_module(
+            "ansible.builtin.command",
+            args=["false"],
+        )
+
+
+def test_ignore_errors_does_not_raise(host):
+    # Use the formal task directive (preferred over the legacy
+    # module_ignore_errors kwarg).
+    result = host.run_module(
+        "ansible.builtin.command",
+        args=["false"],
+        task_directives={"ignore_errors": True},
+    )
+    assert isinstance(result, dict)
+    assert result.get("failed") is True
+    assert result.get("rc") != 0
+
+
+def test_batch_mode_executes_each_task_exactly_once(host):
+    """Regression test for the bug where run_module() in batch mode would
+    queue the task AND immediately execute it (missing `return` in the
+    batch branch). After the fix, batch tasks are only executed at __exit__.
+    """
+    with host:
+        host.run_module("ansible.builtin.ping")
+        host.run_module(
+            "ansible.builtin.command",
+            args=["echo batch-test"],
+        )
+
+    results = host.results
+    # Two tasks queued -> two results back.
+    assert isinstance(results, list), f"expected list, got {type(results).__name__}"
+    assert len(results) == 2, f"expected 2 results, got {len(results)}"
+
+    ping_result, cmd_result = results
+    assert ping_result.get("ping") == "pong"
+    assert cmd_result.get("rc") == 0
+    assert "batch-test" in cmd_result.get("stdout", "")
+
+
+def test_dynamic_dispatch_via_getattr(host):
+    """The __getattr__ shorthand should dispatch to run_module."""
+    result = host.ping()
+    assert isinstance(result, dict)
+    assert result.get("ping") == "pong"
+
+
+def test_dynamic_dispatch_with_positional_arg(host):
+    """`host.command("echo hi")` should pass the positional arg as the
+    module's free-form (_raw_params) input."""
+    result = host.command("echo dynamic-positional")
+    assert isinstance(result, dict)
+    assert result.get("rc") == 0
+    assert "dynamic-positional" in result.get("stdout", "")
+
+
+def test_dynamic_dispatch_with_kwargs(host):
+    """`host.command(cmd=...)` should pass kwargs as the module parameters."""
+    result = host.command(cmd="echo dynamic-kwargs")
+    assert isinstance(result, dict)
+    assert result.get("rc") == 0
+    assert "dynamic-kwargs" in result.get("stdout", "")
+
+
+def test_dynamic_dispatch_inside_with_block(host):
+    """Dynamic dispatch should be deferred too when used inside a with block."""
+    with host:
+        host.ping()
+        host.command("echo from-with-block")
+
+    results = host.results
+    assert isinstance(results, list)
+    assert len(results) == 2
+    assert results[0].get("ping") == "pong"
+    assert "from-with-block" in results[1].get("stdout", "")
+
+
+def test_dynamic_dispatch_passes_task_directives(host):
+    """task_directives should still be honored via dynamic dispatch."""
+    result = host.command("false", task_directives={"ignore_errors": True})
+    assert isinstance(result, dict)
+    assert result.get("failed") is True
+    assert result.get("rc") != 0
+
+
+def test_load_module_then_run_loaded_modules(host):
+    """Explicit load_module + run_loaded_modules sequence (no context manager)."""
+    host.load_module("ansible.builtin.ping")
+    host.load_module(
+        "ansible.builtin.command",
+        args=["echo from-loaded"],
+    )
+
+    results = host.run_loaded_modules()
+    assert isinstance(results, list)
+    assert len(results) == 2
+    assert results[0].get("ping") == "pong"
+    assert "from-loaded" in results[1].get("stdout", "")
+
+
+def test_run_loaded_modules_clears_queue(host):
+    """Successive run_loaded_modules calls must not re-execute earlier tasks."""
+    host.load_module("ansible.builtin.ping")
+    first = host.run_loaded_modules()
+    second = host.run_loaded_modules()
+
+    assert isinstance(first, dict)
+    assert first.get("ping") == "pong"
+    # Second call: queue is empty, returns {} per library contract.
+    assert second == {}
+
+
+def test_unsupported_module_raises(host):
+    """Asking for a nonexistent module should raise UnsupportedAnsibleModule."""
+    from ansible_host import UnsupportedAnsibleModule
+
+    with pytest.raises(UnsupportedAnsibleModule):
+        host.run_module("ansible.builtin.this_module_does_not_exist_xyz")
+
+
+def test_unsupported_module_via_dynamic_dispatch_raises(host):
+    """Same coverage but via the __getattr__ entry point."""
+    from ansible_host import UnsupportedAnsibleModule
+
+    with pytest.raises(UnsupportedAnsibleModule):
+        host.this_module_does_not_exist_xyz()
+
+
+def test_explicit_and_dynamic_dispatch_produce_equivalent_results(host):
+    """Sanity: both invocation styles should yield the same shape of result."""
+    explicit = host.run_module("ansible.builtin.command", args=["echo same"])
+    dynamic = host.command("echo same")
+    assert explicit.get("rc") == dynamic.get("rc") == 0
+    assert "same" in explicit["stdout"]
+    assert "same" in dynamic["stdout"]
+
+
+# -----------------------------------------------------------------------------
+# no_log
+# -----------------------------------------------------------------------------
+
+SECRET = "super-secret-password-xyz-123"
+
+
+def test_no_log_true_censors_invocation_args(host):
+    """With no_log=True, the result must not leak the sensitive cmd string."""
+    result = host.run_module(
+        "ansible.builtin.command",
+        args=[f"echo {SECRET}"],
+        task_directives={"no_log": True},
+    )
+    # The command still ran (stdout is allowed because the user opted into it),
+    # but the args/invocation must be censored so logs don't leak the cmd.
+    serialized = json.dumps(result, default=str)
+    invocation = result.get("invocation") or {}
+    assert SECRET not in json.dumps(invocation, default=str), (
+        "no_log=True must censor invocation args, but SECRET was found in: "
+        f"{invocation!r}"
+    )
+    # Ansible marks censored results explicitly.
+    assert result.get("censored") or result.get("_ansible_no_log") or (
+        "VALUE_SPECIFIED_IN_NO_LOG_PARAMETER" in serialized
+    ), f"no_log=True did not produce a censored result. Full result: {serialized}"
+
+
+def test_no_log_false_includes_invocation_args(host):
+    """Negative control: without no_log, the secret leaks somewhere in the
+    result (stdout from ``echo`` at minimum). This proves the no_log test
+    above is not a false positive (i.e., the secret is genuinely censored
+    rather than just absent from the result for unrelated reasons).
+    """
+    result = host.run_module(
+        "ansible.builtin.command",
+        args=[f"echo {SECRET}"],
+    )
+    serialized = json.dumps(result, default=str)
+    assert SECRET in serialized, (
+        "Without no_log, the secret should be visible somewhere in the result "
+        "(typically stdout). Negative control failed -- the no_log positive "
+        f"test may be passing for the wrong reason. result={result!r}"
+    )
+
+
+# -----------------------------------------------------------------------------
+# forks
+# -----------------------------------------------------------------------------
+
+def test_forks_option_accepted_for_single_host():
+    """forks=N is meaningless for a single host but must not break execution."""
+    host = AnsibleLocalhost(options={"forks": 5})
+    result = host.ping()
+    assert result.get("ping") == "pong"
+
+
+@pytest.fixture
+def multi_host_inventory(tmp_path):
+    """Inventory of 4 'hosts' all backed by the local connection plugin.
+
+    The hostnames are arbitrary labels; ansible_connection=local makes
+    each one execute on the test runner itself. This lets us exercise
+    multi-host fan-out (and the forks knob) without any network/SSH setup.
+    """
+    inv = tmp_path / "hosts.ini"
+    inv.write_text(
+        "[localpool]\n"
+        "node1 ansible_connection=local\n"
+        "node2 ansible_connection=local\n"
+        "node3 ansible_connection=local\n"
+        "node4 ansible_connection=local\n"
+    )
+    return str(inv)
+
+
+def _names_from(results: dict) -> set:
+    return set(results.keys())
+
+
+def test_multi_host_fanout_with_forks_default(multi_host_inventory):
+    """All 4 hosts should produce a result with default forks."""
+
+    hosts = AnsibleHosts(inventory=multi_host_inventory, pattern="all")
+    results = hosts.run_module("ansible.builtin.ping")
+    assert isinstance(results, dict)
+    assert _names_from(results) == {"node1", "node2", "node3", "node4"}
+    for hostname, res in results.items():
+        assert res.get("ping") == "pong", f"{hostname} did not pong: {res!r}"
+
+
+def test_multi_host_fanout_with_forks_one_runs_sequentially(multi_host_inventory):
+    """forks=1 forces sequential execution but must still produce all results."""
+
+    hosts = AnsibleHosts(
+        inventory=multi_host_inventory, pattern="all", options={"forks": 1}
+    )
+    results = hosts.run_module("ansible.builtin.ping")
+    assert _names_from(results) == {"node1", "node2", "node3", "node4"}
+    for res in results.values():
+        assert res.get("ping") == "pong"
+
+
+def test_multi_host_fanout_with_forks_high(multi_host_inventory):
+    """forks=10 (> host count) must not break; all hosts still produce results."""
+
+    hosts = AnsibleHosts(
+        inventory=multi_host_inventory, pattern="all", options={"forks": 10}
+    )
+    results = hosts.run_module("ansible.builtin.ping")
+    assert _names_from(results) == {"node1", "node2", "node3", "node4"}
+    for res in results.values():
+        assert res.get("ping") == "pong"
+
+# ============================================================================
+# AnsibleHosts container protocol + multi-host contract
+# ----------------------------------------------------------------------------
+# Cover the API surface of AnsibleHosts that the fanout tests do NOT exercise:
+# - __len__, __getitem__ (int + str + bad key), __iter__
+# - hosts_count / hostnames properties
+# - pattern matching (subset, empty -> NoAnsibleHostError)
+# - _make_single_host regression (returns AnsibleHost that actually executes)
+# - per-host failure aggregation with ignore_errors
+# - __getattr__ rejects dunder names on AnsibleHosts (parallel to base smoke)
+# ============================================================================
+
+
+def test_hosts_len_count_and_names_agree(multi_host_inventory):
+    hosts = AnsibleHosts(inventory=multi_host_inventory, pattern="all")
+    assert len(hosts) == 4
+    assert hosts.hosts_count == 4
+    assert set(hosts.hostnames) == {"node1", "node2", "node3", "node4"}
+
+
+def test_hosts_pattern_selects_subset(multi_host_inventory):
+    hosts = AnsibleHosts(inventory=multi_host_inventory, pattern="node1:node2")
+    assert len(hosts) == 2
+    assert set(hosts.hostnames) == {"node1", "node2"}
+
+
+def test_hosts_pattern_matching_none_raises(multi_host_inventory):
+    with pytest.raises(NoAnsibleHostError):
+        AnsibleHosts(inventory=multi_host_inventory, pattern="does-not-exist")
+
+
+def test_hosts_getitem_by_int_returns_executable_single_host(multi_host_inventory):
+    """Regression: hosts[0] must reuse parsed inventory AND actually run a module.
+
+    This is the test that pins down the _make_single_host fast-path: if it
+    forgets to copy any field (loader / im / vm / options / _batch_mode / ...),
+    the resulting AnsibleHost will look fine until you try to .run_module()
+    on it.
+    """
+    hosts = AnsibleHosts(inventory=multi_host_inventory, pattern="all")
+    first = hosts[0]
+    assert isinstance(first, AnsibleHost)
+    assert first.hostname == "node1"
+
+    # Single-host result must be unwrapped (no hostname key).
+    result = first.run_module("ansible.builtin.ping")
+    assert isinstance(result, dict)
+    assert result.get("ping") == "pong"
+
+
+def test_hosts_getitem_by_string_matches_int_index(multi_host_inventory):
+    hosts = AnsibleHosts(inventory=multi_host_inventory, pattern="all")
+    by_int = hosts[0]
+    by_str = hosts["node1"]
+    assert by_int.hostname == by_str.hostname == "node1"
+    assert type(by_int) is type(by_str) is AnsibleHost
+
+
+def test_hosts_getitem_bad_keys_raise(multi_host_inventory):
+    hosts = AnsibleHosts(inventory=multi_host_inventory, pattern="all")
+    with pytest.raises(IndexError):
+        hosts[99]
+    with pytest.raises(KeyError):
+        hosts["no-such-node"]
+    with pytest.raises(TypeError):
+        hosts[1.5]  # type: ignore[index]
+
+
+def test_hosts_iteration_yields_one_ansible_host_per_member(multi_host_inventory):
+    hosts = AnsibleHosts(inventory=multi_host_inventory, pattern="all")
+    seen = list(hosts)
+    assert len(seen) == 4
+    assert all(isinstance(h, AnsibleHost) for h in seen)
+    assert [h.hostname for h in seen] == hosts.hostnames
+
+
+def test_hosts_dunder_attribute_does_not_become_module(multi_host_inventory):
+    """AnsibleHostsBase.__getattr__ must NOT treat dunder probes as module names.
+
+    Without this guard, copy/pickle/repr/debuggers probing for things like
+    __deepcopy__ would build an Ansible task named '__deepcopy__' and blow up
+    with UnsupportedAnsibleModule the first time anyone copies a hosts object.
+    """
+    hosts = AnsibleHosts(inventory=multi_host_inventory, pattern="all")
+    with pytest.raises(AttributeError):
+        hosts.__deepcopy__  # noqa: B018  -- attribute access is the test
+
+
+def test_hosts_failure_does_not_drop_others_from_results(multi_host_inventory):
+    """Per-host result aggregation: when run_module fails, every host still
+    appears in the results dict (failures are aggregated, not short-circuited).
+
+    We make all 4 hosts fail (running ``false``) with ignore_errors=True so the
+    call returns instead of raising. The contract being pinned is "the results
+    dict has one entry per matched host even on failure paths" -- the bug this
+    guards against would be silently dropping failed hosts from the dict.
+    """
+    hosts = AnsibleHosts(inventory=multi_host_inventory, pattern="all")
+    results = hosts.run_module(
+        "ansible.builtin.command",
+        args=["false"],
+        task_directives={"ignore_errors": True},
+    )
+    assert isinstance(results, dict)
+    assert set(results.keys()) == {"node1", "node2", "node3", "node4"}
+    for name, res in results.items():
+        assert res.get("failed") is True, f"{name} should be failed: {res!r}"
+        assert res.get("rc") == 1, f"{name} should have rc=1: {res!r}"