Skip to content

fix: parallelize deep_search Tier 3 config fetches (closes #879) #882

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kingpanther13 merged 6 commits into from
Apr 7, 2026
Merged

fix: parallelize deep_search Tier 3 config fetches (closes #879) #882

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ac1ca3b
fix: parallelize deep_search Tier 3 config fetches and remove name-sc…
kingpanther13 Apr 5, 2026
506c6b1
fix: rename gather results to batch_results to avoid shadowing outer …
kingpanther13 Apr 5, 2026
19f4a1e
fix: address review feedback — safe env var parsing, fix budget test …
kingpanther13 Apr 5, 2026
65e3772
Merge branch 'master' into fix/deep-search-tier3-parallel
kingpanther13 Apr 6, 2026
c5e7874
fix: tighten budget test lower bound per review feedback
kingpanther13 Apr 6, 2026
848fbbb
fix: address all 8 review comments from sergeykad
kingpanther13 Apr 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 100 additions & 48 deletions src/ha_mcp/tools/smart_search.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

import asyncio
import logging
import os
import random
import time
from typing import Any
Expand All @@ -23,11 +24,23 @@
BULK_WEBSOCKET_TIMEOUT = 3.0 # Timeout for bulk WebSocket calls
INDIVIDUAL_CONFIG_TIMEOUT = 5.0 # Timeout for individual config fetches

# Time budgets for fallback individual fetching (in seconds)
AUTOMATION_CONFIG_TIME_BUDGET = (
15.0 # Max time for fetching automation configs individually
)
SCRIPT_CONFIG_TIME_BUDGET = 10.0 # Max time for fetching script configs individually
# Time budgets for fallback individual fetching (in seconds).
# Configurable via env vars for instances with many automations/scripts.
def _env_float(key: str, default: float) -> float:
Comment thread
kingpanther13 marked this conversation as resolved.
raw = os.environ.get(key)
if raw is None:
return default
try:
return float(raw)
except (ValueError, TypeError):
logger.warning(f"Invalid value for {key}={raw!r}, using default {default}")
return default

AUTOMATION_CONFIG_TIME_BUDGET = _env_float("HAMCP_AUTOMATION_CONFIG_TIME_BUDGET", 30.0)
SCRIPT_CONFIG_TIME_BUDGET = _env_float("HAMCP_SCRIPT_CONFIG_TIME_BUDGET", 20.0)

# Batch size for parallel individual config fetches (Attempt C fallback)
INDIVIDUAL_FETCH_BATCH_SIZE = 10


def _simplify_states_summary(
Expand Down Expand Up @@ -903,39 +916,61 @@ async def deep_search(
f"Automation WebSocket bulk fetch ({ws_type}) failed: {e}"
)

# Attempt C: Individual REST calls with time budget (LAST RESORT)
# Prioritize name-matched automations so we at least get their configs
# Attempt C: Parallel individual REST calls with time budget (LAST RESORT)
# Fetch configs in parallel batches (subject to time budget) — don't prioritize by name score.
# Name score is only used for result ranking, not fetch order, because
# deep_search's purpose is to find matches INSIDE configs (conditions/actions),
# not just by name. Prioritizing by name would skip the configs most likely
# to contain non-obvious matches. See #879.
if not bulk_fetched:
budget_start = time.perf_counter()
sorted_by_score = sorted(
name_scored, key=lambda x: x[2], reverse=True
)
uids_to_fetch = [
uid
for _, _, _, uid in name_scored
if uid and uid not in all_automation_configs
]
total_to_fetch = len(uids_to_fetch)
fetched_count = 0
failed_count = 0

for (
_entity_id,
_friendly_name,
_name_score,
unique_id,
) in sorted_by_score:
if (
time.perf_counter() - budget_start
> AUTOMATION_CONFIG_TIME_BUDGET
):
break
if not unique_id or unique_id in all_automation_configs:
continue
async def _fetch_automation_config(uid: str) -> tuple[str, dict[str, Any] | None]:
try:
config = await asyncio.wait_for(
self.client._request(
"GET", f"/config/automation/config/{unique_id}"
"GET", f"/config/automation/config/{uid}"
),
timeout=INDIVIDUAL_CONFIG_TIMEOUT,
)
all_automation_configs[unique_id] = config
return (uid, config)
except Exception as e:
logger.debug(
f"Automation individual config fetch ({unique_id}) failed: {e}"
f"Automation individual config fetch ({uid}) failed: {e}"
)
return (uid, None)

for i in range(0, len(uids_to_fetch), INDIVIDUAL_FETCH_BATCH_SIZE):
if (
time.perf_counter() - budget_start
> AUTOMATION_CONFIG_TIME_BUDGET
):
skipped = total_to_fetch - fetched_count - failed_count
logger.warning(
f"Automation config fetch budget exhausted "
f"({AUTOMATION_CONFIG_TIME_BUDGET}s). "
f"Fetched {fetched_count}/{total_to_fetch} "
f"({failed_count} failed), skipped {skipped} automations."
)
break
batch = uids_to_fetch[i : i + INDIVIDUAL_FETCH_BATCH_SIZE]
batch_results = await asyncio.gather(
*[_fetch_automation_config(uid) for uid in batch],
)
for uid_result, config_result in batch_results:
if config_result is not None:
all_automation_configs[uid_result] = config_result
fetched_count += 1
Comment thread
kingpanther13 marked this conversation as resolved.
else:
failed_count += 1

# Phase 3: Score with whatever configs we have
for entity_id, friendly_name, name_score, unique_id in name_scored:
Expand Down Expand Up @@ -1040,37 +1075,54 @@ async def deep_search(
f"Script WebSocket bulk fetch ({ws_type}) failed: {e}"
)

# Attempt C: Individual fetch with budget
# Attempt C: Parallel individual fetch with budget (see #879)
if not script_bulk_fetched:
budget_start = time.perf_counter()
sorted_scripts = sorted(
script_name_scored, key=lambda x: x[3], reverse=True
)
for (
_entity_id,
_friendly_name,
script_id,
_name_score,
) in sorted_scripts:
if (
time.perf_counter() - budget_start
> SCRIPT_CONFIG_TIME_BUDGET
):
break
if script_id in all_script_configs:
continue
sids_to_fetch = [
sid
for _, _, sid, _ in script_name_scored
if sid and sid not in all_script_configs
]
total_to_fetch = len(sids_to_fetch)
fetched_count = 0
failed_count = 0

async def _fetch_script_config(sid: str) -> tuple[str, dict[str, Any] | None]:
try:
config_resp = await asyncio.wait_for(
self.client.get_script_config(script_id),
self.client.get_script_config(sid),
timeout=INDIVIDUAL_CONFIG_TIMEOUT,
)
all_script_configs[script_id] = config_resp.get(
"config", {}
)
return (sid, config_resp.get("config", {}))
except Exception as e:
logger.debug(
f"Script individual config fetch ({script_id}) failed: {e}"
f"Script individual config fetch ({sid}) failed: {e}"
)
return (sid, None)

for i in range(0, len(sids_to_fetch), INDIVIDUAL_FETCH_BATCH_SIZE):
if (
time.perf_counter() - budget_start
> SCRIPT_CONFIG_TIME_BUDGET
):
skipped = total_to_fetch - fetched_count - failed_count
logger.warning(
f"Script config fetch budget exhausted "
f"({SCRIPT_CONFIG_TIME_BUDGET}s). "
f"Fetched {fetched_count}/{total_to_fetch} "
f"({failed_count} failed), skipped {skipped} scripts."
)
break
batch = sids_to_fetch[i : i + INDIVIDUAL_FETCH_BATCH_SIZE]
batch_results = await asyncio.gather(
*[_fetch_script_config(sid) for sid in batch],
)
for sid_result, config_result in batch_results:
if config_result is not None:
all_script_configs[sid_result] = config_result
fetched_count += 1
else:
failed_count += 1

# Phase 3: Score scripts
for (
Expand Down
Loading
Loading