fix(ansible): overlapping runs: one per host, one per schedule, one action run per alert rule (#5158)

- One run per host: start_run claims every host it touches (addresses read
  from the inventory Steward wrote; a source's own inventory file is claimed
  as that file) before it takes a run slot. The claim is all-or-nothing. A run
  waiting on a host shows queued and names the run it waits for, honours a
  cancel, and fails after HOST_WAIT_SECONDS (1 h) with which host and run.
- A due schedule whose previous run is still queued or running skips the
  firing: last_run_at moves on, last_run_id still names the run going, and
  last_error says "Skipped: the previous run was still going". The Schedules
  page shows that beside the run (muted), and a launch error in red.
- An alert rule that fires while the action run from an earlier firing is
  still going starts no new run; the event says so and names the run. An
  in-process set closes the gap between the check and the run being linked.
- docs/core/ansible.md: queued/cancelled statuses and the overlap rules.

Tests: tests/core/test_ansible_overlap.py (claim keys, waiting, all-or-nothing,
cancel, deadline) and tests/integration/test_ansible_overlap.py (two real
runs on one host take turns; schedule skip then fire; alert action skip then
start).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

steward/ansible/executor.py

@@ -38,6 +38,111 @@ def artifact_path(run_id: str) -> str:
     return os.path.join(ARTIFACT_DIR, f"{run_id}.log")
 
 
+# ── One run per host (#5158) ──────────────────────────────────────────────────
+# Two runs against the same host would race each other's changes, so a run
+# claims every host it touches before it takes a run slot, and waits (queued)
+# while another run holds one. The claim is all-or-nothing, so two runs over
+# overlapping hosts can't each hold half and wait on the other.
+
+HOST_WAIT_SECONDS = 3600.0  # a run waiting longer than this for a host fails
+
+
+class HostBusy(Exception):
+    """A run gave up waiting for a host another run held."""
+
+
+def run_hosts(inventory_content: str | None, inventory_path: str | None) -> set[str]:
+    """What a run touches, as claim keys: the address of each host in the YAML
+    inventory Steward wrote for it, or, for a source's own inventory file
+    (whose hosts Steward doesn't read), that file."""
+    if not inventory_content:
+        return {f"inventory:{inventory_path}"} if inventory_path else set()
+    import yaml
+    try:
+        data = yaml.safe_load(inventory_content)
+    except yaml.YAMLError:
+        return set()  # ansible-playbook will refuse it and say why
+    # A host listed again under a group carries no vars there, so its address
+    # is whichever listing gives one.
+    addresses: dict[str, str | None] = {}
+
+    def walk(group) -> None:
+        if not isinstance(group, dict):
+            return
+        members = group.get("hosts")
+        if isinstance(members, dict):
+            for name, hostvars in members.items():
+                addr = hostvars.get("ansible_host") if isinstance(hostvars, dict) else None
+                addresses[str(name)] = addr or addresses.get(str(name))
+        children = group.get("children")
+        if isinstance(children, dict):
+            for child in children.values():
+                walk(child)
+
+    if isinstance(data, dict):
+        for group in data.values():
+            walk(group)
+    return {str(addr or name).strip().lower() for name, addr in addresses.items()}
+
+
+def _host_gate(app: "Quart") -> dict:
+    """The claims and the wake-up event, bound to the running loop like the
+    run-slot semaphore (one loop in prod, one per test)."""
+    loop = asyncio.get_running_loop()
+    gate = getattr(app, "_ansible_host_gate", None)
+    if gate is None or gate["loop"] is not loop:
+        gate = {"loop": loop, "held": {}, "changed": asyncio.Event()}
+        app._ansible_host_gate = gate
+    return gate
+
+
+async def _claim_hosts(app: "Quart", run_id: str, hosts: set[str]) -> bool:
+    """Hold every host in hosts for run_id, waiting while another run holds
+    any of them. Returns whether it had to wait. Raises _Aborted when the run
+    is cancelled while waiting, HostBusy after HOST_WAIT_SECONDS."""
+    from steward.models.ansible import AnsibleRunStatus
+
+    gate = _host_gate(app)
+    held: dict[str, str] = gate["held"]
+    give_up = time.monotonic() + HOST_WAIT_SECONDS
+    waited = False
+    while True:
+        busy = sorted({held[h] for h in hosts if h in held and held[h] != run_id})
+        if not busy:
+            for h in hosts:
+                held[h] = run_id
+            return waited
+        if run_id in _cancelled:
+            raise _Aborted
+        remaining = give_up - time.monotonic()
+        if remaining <= 0:
+            names = ", ".join(sorted(h for h in hosts if h in held))
+            raise HostBusy(f"gave up after {HOST_WAIT_SECONDS:.0f} s waiting for "
+                           f"{names}, held by run {', '.join(busy)}")
+        if not waited:
+            waited = True
+            await _set_status(app, run_id, AnsibleRunStatus.queued)
+            _broadcast(run_id, f"[queued — waiting for run {', '.join(busy)} "
+                               "to finish on the same host]")
+        changed = gate["changed"]
+        try:
+            # Woken by a release; the short timeout also notices a cancel.
+            await asyncio.wait_for(changed.wait(), timeout=min(remaining, 2.0))
+        except asyncio.TimeoutError:
+            pass
+
+
+def _release_hosts(app: "Quart", run_id: str, hosts: set[str]) -> None:
+    gate = getattr(app, "_ansible_host_gate", None)
+    if gate is None:
+        return
+    for h in hosts:
+        if gate["held"].get(h) == run_id:
+            del gate["held"][h]
+    changed, gate["changed"] = gate["changed"], asyncio.Event()
+    changed.set()
+
+
 @dataclass
 class _Done:
     status: str
@@ -392,8 +497,9 @@ async def start_run(
 ) -> None:
     """Execute ansible-playbook as a subprocess and update the DB run row.
 
-    Respects a global concurrency cap (app._ansible_semaphore): if no slot is
-    free the run shows as 'queued' until one frees. Supports cancellation, a
+    Waits (as 'queued') while another run holds one of its hosts (#5158), then
+    for a slot under the global concurrency cap (app._ansible_semaphore).
+    Supports cancellation, a
     persistent full-log artifact, and a parsed per-host result summary.
 
     connection is an optional per-run SSH override (user/password) for
@@ -442,17 +548,23 @@ async def start_run(
         sem = asyncio.Semaphore(max(1, cap))
         app._ansible_semaphore = sem
         app._ansible_sem_loop = loop
-    queued = sem.locked()
-    if queued:
-        await _set_status(app, run_id, AnsibleRunStatus.queued)
-        _broadcast(run_id, "[queued — waiting for an Ansible run slot]")
-    await sem.acquire()
+    hosts = run_hosts(inventory_content, inventory_path)
 
     final_status = "failed"
     logf = None
     proc = None
     interrupted = False
+    slot_held = False
     try:
+        # The hosts first, so a run waiting on a host doesn't sit on a run slot.
+        queued = await _claim_hosts(app, run_id, hosts)
+        if sem.locked():
+            if not queued:
+                await _set_status(app, run_id, AnsibleRunStatus.queued)
+            queued = True
+            _broadcast(run_id, "[queued — waiting for an Ansible run slot]")
+        await sem.acquire()
+        slot_held = True
         if run_id in _cancelled:  # cancelled while queued — abort before launching
             raise _Aborted
         if queued:
@@ -592,7 +704,9 @@ async def start_run(
             logf.close()
         _run_procs.pop(run_id, None)
         _cancelled.discard(run_id)
-        sem.release()
+        if slot_held:
+            sem.release()
+        _release_hosts(app, run_id, hosts)
 
     results = {"hosts": parse_recap(_run_lines.get(run_id, [])), "failures": failures}
     try: