feat(dashboard): serve real recommendations in Nodes & Health; give mesh_intelligence one home (#148)

* refactor(mesh_reporter): expose recommendations as list[str]

Add recommendations_list(scope, scope_value) as the canonical source of
recommendation text. build_recommendations() now just joins that list with
the historical "OPTIMIZATION RECOMMENDATIONS:" header/bullet format it
always used.

router.py:1145 injects build_recommendations()'s output into the LLM system
prompt on the live mesh-DM path — verified byte-identical before/after
across mesh/region/node/missing/empty scopes via a synthetic fixture, and
pinned with a literal-string regression test so a future refactor can't
silently change that prompt text.

This unblocks wiring recommendations into the dashboard, which previously
only reached mesh DMs.

* feat(dashboard): serve real recommendations from /api/health

mesh_reporter was never on app.state — add it alongside health_engine etc.
in server.py's existing pattern. mesh_routes.py's health endpoint now
returns mesh_reporter.recommendations_list("mesh") instead of a hardcoded
[] TODO stub.

Add recommendations_available: bool alongside recommendations: string[] so
"the engine ran and found nothing" (healthy mesh) is never indistinguishable
from "the engine couldn't run" (unwired reporter, or an exception — logged
via logger.exception and swallowed so the rest of the health response still
serves). A crashed recommendations engine must not read as "mesh is
healthy" to an operator.

Also delete main.py's phantom `getattr(mh, "recommendations", [])` on the
websocket health_update push: nothing ever set .recommendations on the
mesh_health object (always []), and no frontend consumer reads it — the
REST endpoint above is the supported path.

* feat(dashboard-frontend): render recommendations in Nodes & Health

Meshtastic → Nodes & Health → Health tab used to be a duplicate
mesh_intelligence config editor (the same MeshIntelligenceSection also
edited from Settings → Intelligence, a stale-tab-overwrite hazard). Replace
it with a real operational view: fetch /api/health and render its
recommendations list.

Three states, matching visual conventions from Dashboard.tsx's alerts list
and MeshCoreRouting.tsx's cross-link note:
- recommendations present → Lightbulb list, one card per item
- empty + recommendations_available → "No recommendations — mesh is
  healthy" (CheckCircle)
- recommendations_available === false → amber warning, not the healthy
  state (AlertTriangle) — a crashed backend must not look healthy

mesh_intelligence config now has exactly one home: Settings. The Health tab
points there via the existing /config?section= deep-link pattern instead of
duplicating the editor. Config.tsx itself is untouched.

MeshHealth.recommendations/.recommendations_available are optional in the
type: the websocket health_update push doesn't include them (REST-only).

---------

Co-authored-by: Matt Johnson <mj@k7zvx.com>
This commit is contained in:
malice 2026-07-17 14:06:15 -06:00 committed by GitHub
commit d7913fddaa
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 432 additions and 102 deletions

View file

@ -32,7 +32,13 @@ export interface MeshHealth {
total_regions: number total_regions: number
unlocated_count: number unlocated_count: number
last_computed: string last_computed: string
recommendations: string[] // Only populated by the REST /api/health response; the websocket
// health_update push includes neither of these (see mesh_routes.py).
recommendations?: string[]
// false when the recommendations engine couldn't run (unwired reporter or
// an exception) — distinct from a genuinely empty `recommendations` list.
// Must NOT be treated the same as "mesh is healthy".
recommendations_available?: boolean
} }
export interface NodeInfo { export interface NodeInfo {

View file

@ -1,11 +1,9 @@
import { useState, useEffect, useCallback } from 'react' import { useState, useEffect, useCallback } from 'react'
import { Save, RotateCcw, RefreshCw, Check } from 'lucide-react' import { Link } from 'react-router-dom'
import { RefreshCw, Lightbulb, CheckCircle, AlertTriangle, ExternalLink } from 'lucide-react'
import Mesh from './Mesh' import Mesh from './Mesh'
import MeshtasticSources from './MeshtasticSources' import MeshtasticSources from './MeshtasticSources'
import { MeshIntelligenceSection, type MeshIntelligenceConfig } from './Config' import { fetchHealth, type MeshHealth } from '@/lib/api'
import { fetchConfig as apiFetchConfig, updateConfig as apiUpdateConfig } from '@/lib/api'
import { useDirty } from '@/context/DirtyContext'
import { notifyRestartRequired } from '@/components/RestartBanner'
const TABS = [ const TABS = [
{ key: 'nodes', label: 'Nodes' }, { key: 'nodes', label: 'Nodes' },
@ -19,14 +17,9 @@ export default function MeshtasticNodes() {
const [activeTab, setActiveTab] = useState<TabKey>('nodes') const [activeTab, setActiveTab] = useState<TabKey>('nodes')
// Health tab state // Health tab state
const { setDirty } = useDirty() const [health, setHealth] = useState<MeshHealth | null>(null)
const [intelligence, setIntelligence] = useState<MeshIntelligenceConfig | null>(null)
const [originalIntelligence, setOriginalIntelligence] = useState<MeshIntelligenceConfig | null>(null)
const [loading, setLoading] = useState(false) const [loading, setLoading] = useState(false)
const [saving, setSaving] = useState(false)
const [error, setError] = useState<string | null>(null) const [error, setError] = useState<string | null>(null)
const [success, setSuccess] = useState<string | null>(null)
const [hasChanges, setHasChanges] = useState(false)
useEffect(() => { useEffect(() => {
document.title = 'Nodes & Health - MeshAI' document.title = 'Nodes & Health - MeshAI'
@ -36,12 +29,10 @@ export default function MeshtasticNodes() {
setLoading(true) setLoading(true)
setError(null) setError(null)
try { try {
const data = (await apiFetchConfig('mesh_intelligence')) as MeshIntelligenceConfig const data = await fetchHealth()
setIntelligence(data) setHealth(data)
setOriginalIntelligence(JSON.parse(JSON.stringify(data)))
setHasChanges(false)
} catch (err) { } catch (err) {
setError(err instanceof Error ? err.message : 'Failed to load mesh intelligence config') setError(err instanceof Error ? err.message : 'Failed to load mesh health')
} finally { } finally {
setLoading(false) setLoading(false)
} }
@ -49,46 +40,10 @@ export default function MeshtasticNodes() {
// Load when Health tab becomes active // Load when Health tab becomes active
useEffect(() => { useEffect(() => {
if (activeTab === 'health' && intelligence === null && !loading) { if (activeTab === 'health' && health === null && !loading) {
fetchData() fetchData()
} }
}, [activeTab, intelligence, loading, fetchData]) }, [activeTab, health, loading, fetchData])
useEffect(() => {
if (intelligence && originalIntelligence) {
setHasChanges(JSON.stringify(intelligence) !== JSON.stringify(originalIntelligence))
}
}, [intelligence, originalIntelligence])
useEffect(() => {
setDirty(hasChanges)
return () => setDirty(false)
}, [hasChanges, setDirty])
const saveConfig = async () => {
if (!intelligence) return
setSaving(true)
setError(null)
setSuccess(null)
try {
const result = await apiUpdateConfig('mesh_intelligence', intelligence)
setOriginalIntelligence(JSON.parse(JSON.stringify(intelligence)))
setHasChanges(false)
setDirty(false)
setSuccess('Mesh intelligence saved successfully')
if (result.restart_required) notifyRestartRequired([])
setTimeout(() => setSuccess(null), 3000)
} catch (err) {
setError(err instanceof Error ? err.message : 'Save failed')
} finally {
setSaving(false)
}
}
const discardChanges = () => {
if (originalIntelligence) setIntelligence(JSON.parse(JSON.stringify(originalIntelligence)))
setHasChanges(false)
}
return ( return (
<div className="space-y-4"> <div className="space-y-4">
@ -113,64 +68,78 @@ export default function MeshtasticNodes() {
{activeTab === 'nodes' && <Mesh />} {activeTab === 'nodes' && <Mesh />}
{activeTab === 'sources' && <MeshtasticSources />} {activeTab === 'sources' && <MeshtasticSources />}
{activeTab === 'health' && ( {activeTab === 'health' && (
<div className="max-w-2xl mx-auto space-y-6"> <div className="max-w-2xl mx-auto space-y-4">
{/* Save bar */} {/* Header */}
<div className="flex items-center justify-between"> <div className="flex items-center justify-between">
<div> <p className="text-sm text-slate-500">
<p className="text-sm text-slate-500"> Optimization recommendations generated from current mesh health.
Mesh health scoring, region management, and automated alerting. </p>
</p> <button
</div> onClick={fetchData}
<div className="flex items-center gap-2"> className="p-2 text-slate-400 hover:text-slate-200 hover:bg-bg-hover rounded transition-colors"
<button title="Refresh"
onClick={fetchData} >
className="p-2 text-slate-400 hover:text-slate-200 hover:bg-bg-hover rounded transition-colors" <RefreshCw size={18} />
title="Refresh" </button>
>
<RefreshCw size={18} />
</button>
<button
onClick={discardChanges}
disabled={!hasChanges}
className="flex items-center gap-2 px-3 py-2 text-slate-400 hover:text-slate-200 disabled:opacity-50 disabled:cursor-not-allowed transition-colors"
>
<RotateCcw size={16} />
Discard
</button>
<button
onClick={saveConfig}
disabled={saving || !hasChanges}
className="flex items-center gap-2 px-4 py-2 bg-accent hover:bg-accent/80 disabled:bg-slate-700 disabled:cursor-not-allowed rounded text-white transition-colors"
>
<Save size={16} />
{saving ? 'Saving...' : 'Save'}
</button>
</div>
</div> </div>
{/* Status messages */} {/* Status messages */}
{error && ( {error && (
<div className="p-3 text-sm bg-red-500/10 text-red-400 border border-red-500/20">{error}</div> <div className="p-3 text-sm bg-red-500/10 text-red-400 border border-red-500/20">{error}</div>
)} )}
{success && (
<div className="p-3 text-sm bg-green-500/10 text-green-400 border border-green-500/20">
<Check size={14} className="inline mr-2" />{success}
</div>
)}
{loading ? ( {loading ? (
<div className="flex items-center justify-center h-32"> <div className="flex items-center justify-center h-32">
<div className="text-slate-400">Loading...</div> <div className="text-slate-400">Loading...</div>
</div> </div>
) : intelligence ? ( ) : health ? (
<div className="bg-bg-card border border-border p-6"> <div className="bg-bg-card border border-border p-4">
<MeshIntelligenceSection data={intelligence} onChange={setIntelligence} /> <h2 className="text-[10px] font-sans uppercase tracking-widest text-[#666] mb-3">
Optimization Recommendations
</h2>
{health.recommendations && health.recommendations.length > 0 ? (
<div className="space-y-2">
{health.recommendations.map((rec, i) => (
<div
key={i}
className="p-3 bg-accent/5 border-l-2 border-accent flex items-start gap-3"
>
<Lightbulb size={16} className="text-accent flex-shrink-0 mt-0.5" />
<span className="text-sm font-sans text-[#e0e0e0]">{rec}</span>
</div>
))}
</div>
) : health.recommendations_available === false ? (
<div className="flex items-center gap-2 text-amber-400 py-4">
<AlertTriangle size={16} />
<span className="font-sans">
Couldn't compute recommendations right now. Check server logs.
</span>
</div>
) : (
<div className="flex items-center gap-2 text-[#777] py-4">
<CheckCircle size={16} className="text-green-500" />
<span className="font-sans">No recommendations mesh is healthy.</span>
</div>
)}
</div> </div>
) : ( ) : (
<div className="flex items-center justify-center h-32"> <div className="flex items-center justify-center h-32">
<div className="text-red-400">Failed to load config</div> <div className="text-red-400">Failed to load mesh health</div>
</div> </div>
)} )}
{/* Cross-link note */}
<div className="flex items-start gap-2 p-3 bg-[#0a0e17] border border-[#1e2a3a] rounded text-sm text-slate-400">
<ExternalLink size={16} className="text-accent mt-0.5 flex-shrink-0" />
<div>
Health scoring, region management, and alerting thresholds are configured on{' '}
<Link to="/config?section=mesh_intelligence" className="text-accent hover:underline">
Settings &rarr; Intelligence
</Link>
.
</div>
</div>
</div> </div>
)} )}
</div> </div>

View file

@ -1,11 +1,13 @@
"""Mesh health and node API routes.""" """Mesh health and node API routes."""
import logging
from datetime import datetime from datetime import datetime
from typing import Optional from typing import Optional
from fastapi import APIRouter, HTTPException, Request from fastapi import APIRouter, HTTPException, Request
router = APIRouter(tags=["mesh"]) router = APIRouter(tags=["mesh"])
logger = logging.getLogger(__name__)
def _serialize_health_score(score) -> dict: def _serialize_health_score(score) -> dict:
@ -68,6 +70,22 @@ async def get_health(request: Request):
health = health_engine.mesh_health health = health_engine.mesh_health
score = health.score score = health.score
# `recommendations_available` distinguishes "the engine ran and found
# nothing" (empty list, mesh is genuinely healthy) from "the engine
# couldn't run" (unwired reporter or an exception) — the two must not
# look identical to the operator. See mesh_reporter.recommendations_list().
mesh_reporter = getattr(request.app.state, "mesh_reporter", None)
recommendations: list[str] = []
recommendations_available = True
if mesh_reporter:
try:
recommendations = mesh_reporter.recommendations_list("mesh")
except Exception:
logger.exception("mesh_reporter.recommendations_list failed")
recommendations_available = False
else:
recommendations_available = False
return { return {
"score": round(score.composite, 1), "score": round(score.composite, 1),
"tier": score.tier, "tier": score.tier,
@ -90,7 +108,8 @@ async def get_health(request: Request):
"total_regions": health.total_regions, "total_regions": health.total_regions,
"unlocated_count": len(health.unlocated_nodes), "unlocated_count": len(health.unlocated_nodes),
"last_computed": _format_timestamp(health.last_computed), "last_computed": _format_timestamp(health.last_computed),
"recommendations": [], # TODO: Add recommendations "recommendations": recommendations,
"recommendations_available": recommendations_available,
} }

View file

@ -132,6 +132,7 @@ async def start_dashboard(meshai_instance: "MeshAI") -> DashboardBroadcaster:
app.state.config_path = meshai_instance.config._config_path app.state.config_path = meshai_instance.config._config_path
app.state.data_store = meshai_instance.data_store app.state.data_store = meshai_instance.data_store
app.state.health_engine = meshai_instance.health_engine app.state.health_engine = meshai_instance.health_engine
app.state.mesh_reporter = getattr(meshai_instance, "mesh_reporter", None)
app.state.alert_engine = getattr(meshai_instance, "alert_engine", None) app.state.alert_engine = getattr(meshai_instance, "alert_engine", None)
app.state.env_store = getattr(meshai_instance, "env_store", None) app.state.env_store = getattr(meshai_instance, "env_store", None)
app.state.notification_router = getattr(meshai_instance, "notification_router", None) app.state.notification_router = getattr(meshai_instance, "notification_router", None)

View file

@ -183,7 +183,6 @@ class MeshAI:
"total_regions": mh.total_regions, "total_regions": mh.total_regions,
"unlocated_count": getattr(mh, "unlocated_count", 0), "unlocated_count": getattr(mh, "unlocated_count", 0),
"last_computed": mh.last_computed, "last_computed": mh.last_computed,
"recommendations": getattr(mh, "recommendations", []),
} }
await self.broadcaster.broadcast("health_update", health_dict) await self.broadcaster.broadcast("health_update", health_dict)
except Exception as e: except Exception as e:

View file

@ -1324,11 +1324,17 @@ class MeshReporter:
return recs return recs
def build_recommendations(self, scope: str, scope_value: str = None) -> str: def recommendations_list(self, scope: str, scope_value: str = None) -> list[str]:
"""Generate actionable optimization recommendations.""" """Generate actionable optimization recommendations as a plain list.
This is the canonical source of recommendation text, consumed both
by build_recommendations() (LLM prompt injection, formatted as a
string) and by API/UI callers that want a list[str] directly (e.g.
the dashboard health endpoint).
"""
health = self.health_engine.mesh_health health = self.health_engine.mesh_health
if not health: if not health:
return "" return []
recs = [] recs = []
@ -1345,11 +1351,17 @@ class MeshReporter:
else: # mesh scope else: # mesh scope
recs.extend(self._mesh_recommendations(health)) recs.extend(self._mesh_recommendations(health))
return recs[:10]
def build_recommendations(self, scope: str, scope_value: str = None) -> str:
"""Generate actionable optimization recommendations."""
recs = self.recommendations_list(scope, scope_value)
if not recs: if not recs:
return "" return ""
lines = ["OPTIMIZATION RECOMMENDATIONS:"] lines = ["OPTIMIZATION RECOMMENDATIONS:"]
for rec in recs[:10]: for rec in recs:
lines.append(f" - {rec}") lines.append(f" - {rec}")
return "\n".join(lines) return "\n".join(lines)

View file

@ -0,0 +1,136 @@
"""API tests for GET /api/health's `recommendations` field.
Covers the dashboard-recommendations wiring: mesh_reporter is exposed on
app.state (mirroring the existing health_engine/data_store/etc. pattern in
dashboard/server.py) and mesh_routes.py's health endpoint now returns real
recommendations from MeshReporter.recommendations_list("mesh") instead of
the old hardcoded `[]`.
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from meshai.dashboard.api.mesh_routes import router
from meshai.mesh_health import HealthScore, MeshHealth
def _client(health_engine, mesh_reporter=None):
app = FastAPI()
app.include_router(router, prefix="/api")
app.state.health_engine = health_engine
app.state.mesh_reporter = mesh_reporter
return TestClient(app)
def _health_engine(mesh_health):
engine = MagicMock()
engine.mesh_health = mesh_health
return engine
def test_health_endpoint_returns_recommendations():
"""recommendations_list("mesh") output reaches the REST /api/health body,
flagged as `recommendations_available: True` (engine ran successfully)."""
mesh_health = MeshHealth(score=HealthScore())
engine = _health_engine(mesh_health)
reporter = MagicMock()
reporter.recommendations_list.return_value = [
"Coverage gap in TestRegion: 3 nodes only reach 1 gateway.",
"No MQTT uplinks in TestRegion. Enable on at least one infrastructure node.",
]
client = _client(engine, mesh_reporter=reporter)
r = client.get("/api/health")
assert r.status_code == 200
body = r.json()
assert body["recommendations"] == [
"Coverage gap in TestRegion: 3 nodes only reach 1 gateway.",
"No MQTT uplinks in TestRegion. Enable on at least one infrastructure node.",
]
assert body["recommendations_available"] is True
reporter.recommendations_list.assert_called_once_with("mesh")
def test_health_endpoint_empty_recommendations_is_marked_available():
"""A genuinely healthy mesh: empty list AND recommendations_available=True.
This is the "healthy" state it must be distinguishable from the
error/unwired states below, which also produce an empty list but with
recommendations_available=False.
"""
mesh_health = MeshHealth(score=HealthScore())
engine = _health_engine(mesh_health)
reporter = MagicMock()
reporter.recommendations_list.return_value = []
client = _client(engine, mesh_reporter=reporter)
r = client.get("/api/health")
assert r.status_code == 200
body = r.json()
assert body["recommendations"] == []
assert body["recommendations_available"] is True
def test_health_endpoint_no_mesh_reporter_configured():
"""mesh_reporter can be None (e.g. Meshtastic not configured) — no crash,
but this must NOT be indistinguishable from "healthy": empty list with
recommendations_available=False, not True.
"""
mesh_health = MeshHealth(score=HealthScore())
engine = _health_engine(mesh_health)
client = _client(engine, mesh_reporter=None)
r = client.get("/api/health")
assert r.status_code == 200
body = r.json()
assert body["recommendations"] == []
assert body["recommendations_available"] is False
def test_health_endpoint_recommendations_error_is_swallowed_but_flagged(caplog):
"""A raising mesh_reporter must not break the health endpoint (the other
fields are still useful), but the failure must be (a) logged, so it's
traceable, and (b) surfaced via recommendations_available=False, so the
UI never renders a crashed engine as "mesh is healthy"."""
mesh_health = MeshHealth(score=HealthScore())
engine = _health_engine(mesh_health)
reporter = MagicMock()
reporter.recommendations_list.side_effect = RuntimeError("boom")
client = _client(engine, mesh_reporter=reporter)
with caplog.at_level("ERROR"):
r = client.get("/api/health")
assert r.status_code == 200
body = r.json()
assert body["recommendations"] == []
assert body["recommendations_available"] is False
# The other fields on the response are unaffected by the recommendations
# failure — a 500 must not take down the whole health endpoint.
assert body["score"] == round(HealthScore().composite, 1)
assert body["tier"] == HealthScore().tier
assert any("recommendations_list failed" in rec.message for rec in caplog.records)
def test_health_endpoint_no_health_data_yet():
"""health_engine.mesh_health is None (not computed yet) — unaffected by recommendations wiring."""
engine = _health_engine(None)
reporter = MagicMock()
client = _client(engine, mesh_reporter=reporter)
r = client.get("/api/health")
assert r.status_code == 200
body = r.json()
assert body["message"] == "Health engine not ready"
reporter.recommendations_list.assert_not_called()

View file

@ -0,0 +1,188 @@
"""Tests for MeshReporter's recommendations engine.
recommendations_list() is the canonical source of recommendation text
(list[str]); build_recommendations() formats that same list into the
"OPTIMIZATION RECOMMENDATIONS:\n - ..." string consumed by router.py's LLM
prompt injection (a live production path see router.py ~line 1145). This
file locks in that build_recommendations() is exactly
recommendations_list() joined with the historical header/bullet format, so
future refactors can't silently change the LLM-facing string.
"""
from __future__ import annotations
import time
import pytest
from meshai.mesh_health import HealthScore, MeshHealth, RegionHealth
from meshai.mesh_models import UnifiedNode
from meshai.mesh_reporter import MeshReporter
def _node(node_num, **kw):
defaults = dict(
node_num=node_num,
node_id_hex=f"!{node_num:08x}",
short_name=f"N{node_num}",
long_name=f"Node {node_num}",
last_heard=time.time(),
is_online=True,
)
defaults.update(kw)
return UnifiedNode(**defaults)
class _FakeHealthEngine:
def __init__(self, mesh_health, packet_threshold=500):
self.mesh_health = mesh_health
self.packet_threshold = packet_threshold
self._nodes = mesh_health.nodes if mesh_health else {}
def get_node(self, identifier):
for n in self._nodes.values():
if str(n.node_num) == str(identifier) or n.node_id_hex == identifier or n.short_name == identifier:
return n
return None
class _FakeDataStore:
def __init__(self, avg_gateways=1.5):
self._avg_gateways = avg_gateways
def get_mesh_deliverability(self):
return {"avg_gateways": self._avg_gateways}
@pytest.fixture
def reporter_with_data():
"""A MeshReporter wired to a small synthetic mesh with several
recommendation-triggering conditions across node/region/mesh scopes."""
n1 = _node(
1,
packets_by_type={"POSITION_APP": 500}, # aggressive interval trigger
channel_utilization=40,
air_util_tx=15,
battery_percent=10,
battery_trend="declining",
predicted_depletion_hours=20,
is_infrastructure=True,
uplink_enabled=False,
)
n2 = _node(2, is_online=False, last_heard=time.time() - 7200, is_infrastructure=True, uplink_enabled=True)
n3 = _node(3, avg_gateways=1.0, packets_sent_24h=1000, text_messages_24h=0)
n4 = _node(4, avg_gateways=1.0, packets_sent_24h=1000, text_messages_24h=0)
n5 = _node(5, avg_gateways=1.0, packets_sent_24h=1000, text_messages_24h=0)
n6 = _node(6, battery_percent=5, battery_trend="declining")
n7 = _node(7, channel_utilization=20)
nodes = {n.node_num: n for n in [n1, n2, n3, n4, n5, n6, n7]}
region = RegionHealth(
name="TestRegion",
node_ids=[str(i) for i in range(1, 8)],
score=HealthScore(util_percent=30, infra_total=2, infra_online=1),
)
mesh_health = MeshHealth(regions=[region], nodes=nodes)
engine = _FakeHealthEngine(mesh_health)
return MeshReporter(engine, data_store=_FakeDataStore())
@pytest.mark.parametrize(
"scope,scope_value",
[
("mesh", None),
("region", "TestRegion"),
("node", "1"),
("node", "2"),
("node", "999"), # missing node
("region", "Nowhere"), # missing region
],
)
def test_build_recommendations_matches_recommendations_list(reporter_with_data, scope, scope_value):
"""build_recommendations() must be exactly recommendations_list() formatted
with the historical header + bullet convention (byte-identical LLM prompt
text is the whole point of this refactor)."""
recs = reporter_with_data.recommendations_list(scope, scope_value)
text = reporter_with_data.build_recommendations(scope, scope_value)
if not recs:
assert text == ""
else:
expected_lines = ["OPTIMIZATION RECOMMENDATIONS:"] + [f" - {r}" for r in recs]
assert text == "\n".join(expected_lines)
def test_build_recommendations_llm_string_pinned(reporter_with_data):
"""Pins the exact LLM-facing string router.py:1145 injects into the
system prompt (a live production path) against a literal, hardcoded
expectation independent of the implementation, unlike the
recommendations_list()-derived check above. Captured against
origin/main before the recommendations_list() refactor via a synthetic
fixture identical to reporter_with_data's, and confirmed byte-identical
after. If this test ever needs to change, the LLM prompt text changed
and that must be a deliberate, reviewed decision not a refactor
side-effect.
"""
assert reporter_with_data.build_recommendations("mesh") == (
"OPTIMIZATION RECOMMENDATIONS:\n"
" - Coverage gap in TestRegion: 3 nodes only reach 1 gateway. "
"A new MQTT feeder in this area would add monitoring redundancy.\n"
" - Node 6 (N6) at 5% battery and declining. Likely offline soon.\n"
" - High channel utilization on Node 1 (N1), Node 7 (N7). "
"Check for aggressive broadcast intervals or nearby interference.\n"
" - Mesh-wide average is 1.5 gateways per packet. "
"Adding MQTT feeders would improve monitoring reliability across the mesh."
)
assert reporter_with_data.build_recommendations("region", "TestRegion") == (
"OPTIMIZATION RECOMMENDATIONS:\n"
" - Channel utilization at 30%. Consider spreading nodes across "
"frequencies or reducing telemetry intervals.\n"
" - 1 infrastructure node(s) offline. Check power and connectivity.\n"
" - High-traffic nodes (Node 3 (N3), Node 4 (N4), Node 5 (N5)) "
"impacting channel. Review their telemetry settings.\n"
" - Nodes with frequent position broadcasts (Node 1 (N1)). "
"Recommend 900s interval."
)
assert reporter_with_data.build_recommendations("node", "2") == (
"OPTIMIZATION RECOMMENDATIONS:\n"
" - Node offline since 2h ago. Check power and connectivity."
)
assert reporter_with_data.build_recommendations("node", "999") == ""
assert reporter_with_data.build_recommendations("region", "Nowhere") == ""
def test_recommendations_list_caps_at_ten(reporter_with_data):
recs = reporter_with_data.recommendations_list("node", "1")
assert len(recs) <= 10
def test_recommendations_list_empty_when_no_health_data():
engine = _FakeHealthEngine(None)
reporter = MeshReporter(engine, data_store=_FakeDataStore())
assert reporter.recommendations_list("mesh") == []
assert reporter.build_recommendations("mesh") == ""
def test_recommendations_list_empty_mesh_no_triggers():
"""An empty mesh with no nodes/regions produces no recommendations."""
mesh_health = MeshHealth()
engine = _FakeHealthEngine(mesh_health)
reporter = MeshReporter(engine, data_store=_FakeDataStore(avg_gateways=3.0))
assert reporter.recommendations_list("mesh") == []
assert reporter.build_recommendations("mesh") == ""
def test_recommendations_list_returns_plain_strings(reporter_with_data):
recs = reporter_with_data.recommendations_list("mesh")
assert isinstance(recs, list)
assert all(isinstance(r, str) for r in recs)
# Plain recommendation text, not pre-formatted with the LLM-prompt header
# or bullet markers — that formatting belongs to build_recommendations().
assert all(not r.startswith("OPTIMIZATION RECOMMENDATIONS") for r in recs)
assert all(not r.startswith(" - ") for r in recs)