Skip to content

Commit bd22f83

Browse files
committed
feat(cyt_mcp): enhance workspace tool hydration and background processing
- Introduced a new function `_workspace_scope_disk_tools` to load workspace-scoped disk catalogs, improving tool retrieval logic. - Updated `_hydrate_missing_servers_from_disk` to merge tools from both merged disk and workspace disk sources for better server hydration. - Refactored `_hydrate_server_runtime_caches` to return a boolean indicating hydration success, allowing for conditional cache pushes. - Added a new threaded function `_schedule_hydrate_hook_caches_from_registration` to run hook cache hydration asynchronously, improving performance during catalog registration. - Increased `PUSH_TIMEOUT_SECONDS` to 30 to accommodate larger workspace catalogs during hook daemon operations. - Enhanced unit tests to validate new functionality and ensure resilience against stale tool registrations.
1 parent cb5c2ea commit bd22f83

7 files changed

Lines changed: 161 additions & 11 deletions

File tree

‎src/cyt/cyt_mcp/catalog.py‎

Lines changed: 47 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -380,6 +380,42 @@ def _disk_catalog_tools(cache_key: _CytMcpCacheKey) -> list[dict[str, Any]]:
380380
return _normalize_tools_list(tools)
381381

382382

383+
def _workspace_scope_disk_tools(cache_key: _CytMcpCacheKey) -> list[dict[str, Any]]:
384+
"""Load ws-scoped disk catalog (separate slug from merged hook injection cache)."""
385+
if not cache_key.workspace:
386+
return []
387+
ws = Path(cache_key.workspace).expanduser()
388+
agg_path = ws / ".agents/cyt/config/mcp-config.yaml"
389+
if not agg_path.is_file():
390+
from cyt_mcp.workspace_catalog import workspace_aggregator_path
391+
392+
alt = workspace_aggregator_path(ws, cache_key.agent)
393+
if alt is None or not alt.is_file():
394+
return []
395+
agg_path = alt
396+
try:
397+
from cyt_mcp.catalog_build import disk_catalog_slug_for_config
398+
from cyt_mcp.config import load_aggregator_config
399+
400+
config = load_aggregator_config(
401+
agent=cache_key.agent,
402+
aggregator_path=agg_path,
403+
workspace_folder=ws,
404+
)
405+
if config.catalog_scope != "workspace":
406+
return []
407+
slug = disk_catalog_slug_for_config(config)
408+
except (ValueError, OSError):
409+
return []
410+
envelope = read_disk_catalog(slug)
411+
if envelope is None:
412+
return []
413+
tools = envelope.get("tools")
414+
if not isinstance(tools, list):
415+
return []
416+
return _normalize_tools_list(tools)
417+
418+
383419
def _server_keys_for_tools(tools: Sequence[dict[str, Any]]) -> set[str]:
384420
return {key for key in (_tool_server_key(tool) for tool in tools) if key}
385421

@@ -516,16 +552,23 @@ def _hydrate_missing_servers_from_disk(
516552
cache_key: _CytMcpCacheKey,
517553
tools: list[dict[str, Any]],
518554
) -> list[dict[str, Any]]:
519-
disk_tools = _disk_catalog_tools(cache_key)
520-
if not disk_tools:
555+
merged_disk_tools = _disk_catalog_tools(cache_key)
556+
ws_disk_tools = _workspace_scope_disk_tools(cache_key)
557+
disk_sources: list[list[dict[str, Any]]] = []
558+
if merged_disk_tools:
559+
disk_sources.append(merged_disk_tools)
560+
if ws_disk_tools:
561+
disk_sources.append(ws_disk_tools)
562+
if not disk_sources:
521563
return tools
522-
missing_servers = _server_keys_for_tools(disk_tools) - _server_keys_for_tools(tools)
564+
all_disk_tools = [tool for source in disk_sources for tool in source]
565+
missing_servers = _server_keys_for_tools(all_disk_tools) - _server_keys_for_tools(tools)
523566
configured_servers = _configured_server_keys_for_hydration(cache_key)
524567
if configured_servers:
525568
missing_servers = missing_servers & configured_servers
526569
if not missing_servers:
527570
return tools
528-
merged = _preserve_tools_for_servers(tools, [disk_tools], missing_servers)
571+
merged = _preserve_tools_for_servers(tools, disk_sources, missing_servers)
529572
if len(merged) > len(tools):
530573
logger.info(
531574
"cyt-mcp catalog hydrated %d tools from disk for backends: %s",

‎src/cyt/hook/catalog_registry.py‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -480,6 +480,24 @@ def _hydrate_hook_caches_from_registration(
480480
logger.warning("hook cache hydration after catalog register failed: %s", exc)
481481

482482

483+
def _schedule_hydrate_hook_caches_from_registration(
484+
tools: list[dict[str, Any]],
485+
*,
486+
agent: str,
487+
workspace_root: str,
488+
) -> None:
489+
"""Run hook cache hydration off the catalog register HTTP request thread."""
490+
if not tools:
491+
return
492+
thread = threading.Thread(
493+
target=_hydrate_hook_caches_from_registration,
494+
kwargs={"tools": tools, "agent": agent, "workspace_root": workspace_root},
495+
name="cyt-catalog-register-hydrate",
496+
daemon=True,
497+
)
498+
thread.start()
499+
500+
483501
def _register_full_tools(
484502
existing: _CatalogRegistration | None,
485503
*,
@@ -577,7 +595,7 @@ def register_catalog(payload: dict[str, Any]) -> RegisterResult:
577595
from cyt.hook.active_workspace import touch_active_workspace
578596

579597
touch_active_workspace(agent, workspace_root)
580-
_hydrate_hook_caches_from_registration(
598+
_schedule_hydrate_hook_caches_from_registration(
581599
tools,
582600
agent=agent,
583601
workspace_root=workspace_root,

‎src/cyt_mcp/cli.py‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -246,10 +246,10 @@ def _start_pairing_repair_thread(config: AggregatorConfig) -> None:
246246
def _hydrate_server_runtime_caches(
247247
cache: RuntimeToolCache,
248248
config: AggregatorConfig,
249-
) -> None:
249+
) -> bool:
250250
from cyt_mcp.catalog_build import hydrate_runtime_cache
251251

252-
hydrate_runtime_cache(cache, config)
252+
return hydrate_runtime_cache(cache, config)
253253

254254

255255
def _make_background_catalog_refresh(
@@ -329,7 +329,13 @@ async def _run_server(config: AggregatorConfig, *, aggregator_path: Path | None
329329
from cyt.hook.active_workspace import touch_active_workspace
330330

331331
touch_active_workspace(config.agent, config.workspace_root)
332-
_hydrate_server_runtime_caches(cache, config)
332+
hydrated = _hydrate_server_runtime_caches(cache, config)
333+
if bool(cache.snapshot()):
334+
# Disk/registry hydrate can warm the cache before stdio starts; still push so
335+
# hook inject/tiers see ws tools without waiting for a full backend refresh.
336+
from cyt_mcp.hook_daemon_push import schedule_catalog_push
337+
338+
schedule_catalog_push(cache, config)
333339
background_catalog_refresh = _make_background_catalog_refresh(
334340
cache=cache,
335341
config=config,

‎src/cyt_mcp/hook_daemon_push.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,9 @@
2727

2828
REGISTER_PATH = "/hook/catalog/register"
2929
DEREGISTER_PATH = "/hook/catalog/deregister"
30-
PUSH_TIMEOUT_SECONDS = 2.0
30+
# Full ws catalogs (100+ tools) can exceed 2s because the hook daemon rebuilds
31+
# master caches synchronously during /hook/catalog/register.
32+
PUSH_TIMEOUT_SECONDS = 30.0
3133

3234
_RETRY_DELAYS_SECONDS = (1.0, 2.0, 5.0, 10.0)
3335

‎src/tests/unit/test_cyt_mcp_catalog_resilience.py‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -345,3 +345,73 @@ def test_disk_snapshot_hydrate_restores_ws_tools_after_registry_clear(
345345
assert loaded == 1
346346
merged = catalog_for_hook("cursor", workspace, allow_stale=True)
347347
assert {tool["name"] for tool in merged} == {tool["name"] for tool in tools}
348+
349+
350+
def test_hook_catalog_hydrates_ws_disk_when_merged_disk_is_usr_only(
351+
tmp_path: Path,
352+
monkeypatch: pytest.MonkeyPatch,
353+
) -> None:
354+
"""Regression: stale merged hook disk must not hide ws-layer disk tools."""
355+
from cyt.cyt_mcp.catalog import (
356+
_cache_key_for_config,
357+
_write_catalog_disk,
358+
get_cyt_mcp_catalog,
359+
)
360+
from cyt.cyt_mcp.catalog_disk import raw_catalog_content_hash, write_disk_catalog
361+
from cyt_mcp.catalog_build import disk_catalog_slug_for_config
362+
from cyt_mcp.config import load_aggregator_config
363+
from tests.support.cyt_mcp_catalog_resilience_fixtures import (
364+
RESILIENCE_MCP_SERVER_KEYS,
365+
clear_cyt_mcp_memory_catalog_state,
366+
)
367+
368+
workspace = materialize_workspace(tmp_path)
369+
cyt_config_dir = workspace / ".agents" / "cyt" / "config"
370+
(cyt_config_dir / "mcp-config.yaml").write_text(
371+
"\n".join(
372+
[
373+
"default_agent: cursor",
374+
"agents:",
375+
" cursor: mcp/cursor.json",
376+
"catalog_scope: workspace",
377+
"transport: stdio",
378+
],
379+
),
380+
encoding="utf-8",
381+
)
382+
(cyt_config_dir / "mcp" / "cursor.json").write_text(
383+
json.dumps({"mcpServers": {key: {"command": key} for key in RESILIENCE_MCP_SERVER_KEYS}}),
384+
encoding="utf-8",
385+
)
386+
cache_dir = tmp_path / "cyt-mcp-catalog"
387+
monkeypatch.setattr("cyt.cyt_mcp.catalog_disk.cyt_mcp_catalog_cache_dir", lambda: cache_dir)
388+
389+
ws_tools = load_ws_tools_catalog()
390+
usr_tools = load_usr_tools_catalog()
391+
config = load_aggregator_config(
392+
agent="cursor",
393+
aggregator_path=cyt_config_dir / "mcp-config.yaml",
394+
workspace_folder=workspace,
395+
)
396+
ws_slug = disk_catalog_slug_for_config(config)
397+
assert ws_slug is not None
398+
write_disk_catalog(
399+
ws_slug,
400+
agent="cursor",
401+
tools=ws_tools,
402+
content_hash=raw_catalog_content_hash(ws_tools),
403+
)
404+
405+
hook_config = cyt_mcp_hook_config(workspace, db_path=tmp_path / "tiers.db")
406+
merged_key = _cache_key_for_config(hook_config)
407+
_write_catalog_disk(merged_key, usr_tools)
408+
409+
clear_catalog_registry(purge_disk_snapshot=False)
410+
clear_cyt_mcp_memory_catalog_state()
411+
patch_daemon_catalog_status(monkeypatch, [])
412+
413+
catalog = get_cyt_mcp_catalog(hook_config, blocking=True, force=True)
414+
assert catalog is not None
415+
names = {tool["name"] for tool in catalog}
416+
assert {tool["name"] for tool in ws_tools if tool.get("server_key")}.issubset(names)
417+
assert {tool["name"] for tool in usr_tools}.issubset(names)

‎src/tests/unit/test_cyt_mcp_cli.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,8 +56,9 @@ def test_run_server_stdio_skips_background_refresh_when_cache_warmed() -> None:
5656
config = _stdio_config()
5757
refresh_calls: list[str] = []
5858

59-
def _hydrate_warm(cache: RuntimeToolCache, _config: object) -> None:
59+
def _hydrate_warm(cache: RuntimeToolCache, _config: object) -> bool:
6060
cache.replace([{"name": "tool-a"}])
61+
return True
6162

6263
async def _refresh(*_args: object, **_kwargs: object) -> None:
6364
refresh_calls.append("refresh")
@@ -72,13 +73,15 @@ async def _run_async(*_args: object, **_kwargs: object) -> None:
7273
patch("cyt_mcp.catalog_build.refresh_catalog_cache", side_effect=_refresh),
7374
patch("cyt_mcp.hook_daemon_push.register_push_context"),
7475
patch("cyt_mcp.hook_daemon_push.deregister_catalog_push"),
76+
patch("cyt_mcp.hook_daemon_push.schedule_catalog_push") as schedule_push,
7577
):
7678
server = AsyncMock()
7779
build.return_value = (server, None, _mock_coordinator())
7880
server.run_async = AsyncMock(side_effect=_run_async)
7981
result = asyncio.run(_run_server(config))
8082
assert result == 0
8183
assert refresh_calls == []
84+
schedule_push.assert_called_once()
8285

8386

8487
def test_run_search_wires_refresh_and_lookup(

‎src/tests/unit/test_empty_home_cache_bootstrap.py‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,15 @@ def test_register_catalog_eager_hydrates_master_without_warm(
147147

148148
register_ws_catalog(pack.workspace, pack.tools)
149149

150-
catalog = get_master_tool_catalog(config, blocking=False) or []
150+
import time
151+
152+
deadline = time.monotonic() + 5.0
153+
catalog: list[dict[str, object]] = []
154+
while time.monotonic() < deadline:
155+
catalog = get_master_tool_catalog(config, blocking=False) or []
156+
if len(catalog) >= pack.minimum_master_catalog_tools:
157+
break
158+
time.sleep(0.05)
151159
names = {tool["name"] for tool in catalog}
152160
assert len(catalog) >= pack.minimum_master_catalog_tools
153161
assert set(pack.expected_tool_names).issubset(names)

0 commit comments

Comments
 (0)