feat(alerts): renames carry their alerts — monitor and host renames move rules, windows and history (#5065)

#5065 "Renames carry their alerts". Alert rules, maintenance-window scopes
and metric history key a thing by (source, name), so a rename used to leave
the rules watching a name nothing reported under, and split the history.

- core/renames.carry_rename moves AlertRule.resource_name,
  MaintenanceWindow.scope_resource, plugin_metrics and plugin_metrics_hourly
  to the new name in the caller's transaction. A host rename also carries
  its parts: host_agent "<host>:…" and docker "<host>/…" (HOST_SOURCES).
  Matching is exact (a prefix compare, not LIKE or a collation range).
  History left under the new name by something deleted is dropped first
  rather than merged in.
- Web monitor edit, MCP update_monitor and web host edit call it; MCP
  answers with `carried` (it used to return a warning naming stranded rules).
- A name another monitor of the same type, or another host, already has is
  refused on create and rename: two would share one history.
- The rename is audited (monitor.updated / host.renamed) with what moved.

Tests: an integration test renames a firing monitor through the web route
and finds its rule still firing and cleared by a sample on the new name,
its window and raw+hourly history moved and an orphan's history dropped; a
host rename moves agent and container parts but not a longer-named host's
or another source's; a guard that every host_agent resource is "<host>" or
"<host>:…".

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

steward/core/renames.py

@@ -0,0 +1,82 @@
+"""steward/core/renames.py
+
+Alert rules, maintenance-window scopes and metric history all key a thing
+by (source, name). When a monitor or host is renamed, carry_rename moves
+every one of them to the new name, in the caller's transaction, so the
+thing's rules keep watching it and its history stays in one piece (#5065).
+"""
+from __future__ import annotations
+
+from sqlalchemy import and_, delete, func, or_, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from steward.models.alerts import AlertRule, MaintenanceWindow
+from steward.models.metrics import PluginMetric, PluginMetricHourly
+
+# What a host's name keys, and how each source names a part of the host:
+# host_agent "web01:/var", "web01:core3"; docker "web01/nginx".
+HOST_SOURCES: tuple[tuple[str, str | None], ...] = (("host_agent", ":"), ("docker", "/"))
+
+
+def _matches(source_col, name_col, sources, name: str):
+    """(source, name) itself, or a sub-resource "<name><sep>…" of it."""
+    clauses = []
+    for source, sep in sources:
+        under = [name_col == name]
+        if sep:
+            # An exact prefix: LIKE would read % or _ in a name as wildcards, and
+            # a >=/< range follows the collation, which may ignore punctuation.
+            under.append(func.left(name_col, len(name) + len(sep)) == name + sep)
+        clauses.append(and_(source_col == source, or_(*under)))
+    return or_(*clauses)
+
+
+def _renamed(name_col, old: str, new: str):
+    return func.concat(new, func.substr(name_col, len(old) + 1))
+
+
+async def carry_rename(session: AsyncSession, sources, old: str, new: str) -> dict[str, int]:
+    """Point everything keyed by `old` under each (source, separator) in
+    `sources` at `new`. A separator also carries the sub-resources named
+    "<old><sep>…"; None carries the exact name only.
+
+    The caller has already refused a `new` that a live thing holds, so any
+    history under `new` belongs to something deleted: it is dropped first,
+    rather than merged into this thing's history. Returns rows moved, by kind.
+    """
+    if not old or old == new:
+        return {}
+    sources = tuple(sources)
+    for model in (PluginMetric, PluginMetricHourly):
+        await session.execute(delete(model).where(
+            _matches(model.source_module, model.resource_name, sources, new)))
+    moved = {}
+    for kind, model, src, res in (
+        ("rules", AlertRule, AlertRule.source_module, AlertRule.resource_name),
+        ("maintenance_windows", MaintenanceWindow, MaintenanceWindow.scope_source,
+         MaintenanceWindow.scope_resource),
+        ("metric_rows", PluginMetric, PluginMetric.source_module, PluginMetric.resource_name),
+        ("hourly_rows", PluginMetricHourly, PluginMetricHourly.source_module,
+         PluginMetricHourly.resource_name),
+    ):
+        result = await session.execute(
+            update(model).where(_matches(src, res, sources, old))
+            .values({res: _renamed(res, old, new)})
+            .execution_options(synchronize_session=False))
+        moved[kind] = result.rowcount
+    return moved
+
+
+def describe(moved: dict[str, int]) -> str:
+    """'2 alert rules, 1 maintenance window and 18,204 metric samples', or ''."""
+    parts = []
+    for kind, one, many in (("rules", "alert rule", "alert rules"),
+                            ("maintenance_windows", "maintenance window", "maintenance windows")):
+        if moved.get(kind):
+            parts.append(f"{moved[kind]} {one if moved[kind] == 1 else many}")
+    samples = moved.get("metric_rows", 0) + moved.get("hourly_rows", 0)
+    if samples:
+        parts.append(f"{samples:,} metric sample{'' if samples == 1 else 's'}")
+    if len(parts) > 1:
+        return ", ".join(parts[:-1]) + " and " + parts[-1]
+    return "".join(parts)