diff --git a/api/routes/internal.py b/api/routes/internal.py index 6286d08..e72a4ea 100644 --- a/api/routes/internal.py +++ b/api/routes/internal.py @@ -1,16 +1,25 @@ """Internal API routes — not mounted in OpenAPI schema. -Called by Fly.io cron machines. All endpoints require X-Internal-Token header. +Invoked by an external scheduler (exact mechanism/cadence is deployment +config, not tracked in this repo). All endpoints require X-Internal-Token +header. Neon/idle-spin-down note ------------------------- ``_run_notification_check`` is gated by ``shared.notify_due`` — a Redis next-due cache — so that regardless of how often this endpoint is actually invoked (by whatever scheduler ends up calling it), it costs nothing but a -single Redis range query when no user has any anchor/followup/meeting-event -work due right now. See shared/notify_due.py for the full scheme. This makes -the endpoint safe to call as frequently as desired without keeping managed +single Redis range query when no user has any anchor/followup work due +right now. See shared/notify_due.py for the full scheme. This makes the +endpoint safe to call as frequently as desired without keeping managed Postgres (Neon) awake between real due events. + +Note: this endpoint does NOT drain meeting events. `drain_meeting_events` +reads from an in-process queue owned by the BOT process +(`tether_premium.bot.scheduling.events`, populated by a WS-listener thread +started only inside `bot/message_handler.py`'s `run_polling()`) — this API +process cannot see that queue, so calling it here would always be a no-op. +Meeting-event draining happens exclusively via the bot's polling loop. """ from __future__ import annotations @@ -45,7 +54,8 @@ def _verify_internal_token(request: Request) -> None: @router.post("/notifications/check") async def check_notifications(request: Request, background_tasks: BackgroundTasks): - """Called by Fly.io cron every minute. Queues notification check as BackgroundTask.""" + """Queues a notification check as a BackgroundTask. Gated by shared.notify_due + (see module docstring), so safe to invoke at whatever cadence the caller uses.""" _verify_internal_token(request) pool = request.app.state.pool ws_manager = getattr(request.app.state, "ws_manager", None) @@ -169,41 +179,27 @@ async def _recompute_and_cache_due( Called once per user after a real check has run — self-perpetuating: each real run recomputes its own future due time from data it already fetched, so the cache never needs a separate background refresh job. - The "meeting" component is contributed independently by tether-premium's - drain_meeting_events via the same shared.notify_due.set_component_due — - this function only owns the anchor/followup components. + A ``None`` estimate means "nothing to re-check on a timer for this + component right now" — left unset (absent from the hash) rather than + writing a synthetic "never" value, so a future write (e.g. an anchor + create/update, or the next anchor boundary re-triggering follow-ups) + naturally repopulates and recomputes the combined score. """ - now_ts = time.time() if anchor_next is not None: await notify_due.set_component_due(user_id, "anchor", anchor_next.timestamp()) - else: - # No anchors configured — nothing to re-check on a timer for this - # component; leave it unset (absent from the hash) rather than - # writing a synthetic "never" value, since a future anchor - # create/update will populate it and recompute the combined score. - pass if followup_next is not None: await notify_due.set_component_due(user_id, "followup", followup_next.timestamp()) - else: - # No active followup rows right now — the anchor component's next - # boundary (which triggers _init_followup_states) is what will next - # wake this user for the followup side too. - pass - if anchor_next is None and followup_next is None: - logger.debug( - "notify_due: no anchor or followup component for user_id=%s — " - "relying on existing cache entries / fail-open on next check", - user_id, - ) - _ = now_ts # reserved for future observability (e.g. staleness metrics) async def _run_notification_check(pool, ws_manager) -> None: - """For each linked user: check anchor transitions, follow-ups, meeting events. + """For each linked user: check anchor transitions and follow-ups. Gated by shared.notify_due: a single Redis range query decides which users (if any) have real work due right now. When nothing is due, this function returns without touching Postgres at all — see module docstring. + + Does not drain meeting events — see the module docstring's note on why + that would always be a no-op in this (API) process. """ if pool is None: logger.warning("_run_notification_check: no pool available, skipping") @@ -243,14 +239,13 @@ async def _run_notification_check(pool, ws_manager) -> None: for user in users: user_id = str(user["id"]) - anchor_next: datetime | None = None - try: - dispatch_fn = functools.partial( - notify.dispatch, pool=pool, ws_manager=ws_manager - ) - anchor_next = await _check_anchor_transitions(pool, user_id, dispatch_fn) - except Exception as e: - logger.warning("Anchor check failed for user %s: %s", user_id, e) + # No try/except here: _check_anchor_transitions already catches and + # logs everything internally (returning None on failure) — an outer + # catch here could never fire and was dead code. + dispatch_fn = functools.partial( + notify.dispatch, pool=pool, ws_manager=ws_manager + ) + anchor_next = await _check_anchor_transitions(pool, user_id, dispatch_fn) followup_next: datetime | None = None try: @@ -269,28 +264,6 @@ async def _notify_send(text, uid=user_id): except Exception as e: logger.warning("Followup check failed for user %s: %s", user_id, e) - try: - from tether_premium.bot.scheduling.events import drain_meeting_events - - async def _notify_meeting_send(text, uid=user_id): - await notify.dispatch( - uid, - "meeting_event", - text, - pool, - priority="important", - ws_manager=ws_manager, - ) - - meeting_send = lambda text, uid=user_id: asyncio.ensure_future( - _notify_meeting_send(text, uid) - ) - await drain_meeting_events(pool=pool, user_id=user_id, send_fn=meeting_send) - except ImportError: - pass - except Exception as e: - logger.warning("Meeting event drain failed for user %s: %s", user_id, e) - await _recompute_and_cache_due(user_id, anchor_next, followup_next) diff --git a/shared/notify_due.py b/shared/notify_due.py index 190b4a1..94fff9f 100644 --- a/shared/notify_due.py +++ b/shared/notify_due.py @@ -1,4 +1,4 @@ -"""Redis next-due gating for notification checks (anchors/followups/meetings). +"""Redis next-due gating for notification checks (anchors/followups). Purpose ------- @@ -20,19 +20,19 @@ (``get_due_user_ids``) — done by the cron-style entry point — sees a user as due if ANY one component says so, regardless of which side wrote it. -Meeting events are NOT modeled as a component here. They're PUSH-based -(delivered via an in-process WS-listener queue owned by tether-premium's +Meeting events are NOT modeled as a component here, and never have been +written by any caller. They're PUSH-based (delivered via an in-process +WS-listener queue owned by tether-premium's ``tether_premium.bot.scheduling.events``, at unpredictable times) rather than time-based, so there's no meaningful "next_due timestamp" to precompute for them — they don't fit this module's timestamp-gating model. ``drain_meeting_events`` already self-gates for free (it checks its queue is non-empty before ever touching Postgres) and is deliberately called UNCONDITIONALLY by ``run_polling``, exempt from the ``is_due()`` gate here -— see the call site in bot/message_handler.py for the reasoning. The -``"meeting"`` value below is reserved in the ``Component`` type for -possible future use (e.g. if the event queue is ever moved to a shared/ -cross-process backing store), but nothing currently writes it, and none -of the gates in this repo currently depend on it. +— see the call site in bot/message_handler.py for the reasoning. If meeting +events ever need cross-process draining (e.g. the event queue moving to a +shared backing store), that redesign should introduce its own mechanism +rather than reusing this module's timestamp model. Redis layout ------------ @@ -85,7 +85,7 @@ # back to the "unknown user" fail-open path rather than staying wrong forever. _SAFETY_TTL_SECONDS = 24 * 3600 -Component = Literal["anchor", "followup", "meeting"] +Component = Literal["anchor", "followup"] def get_redis_url() -> str | None: @@ -143,12 +143,24 @@ def log_startup_status() -> None: _warn_no_redis_configured_once() +_client_cache: dict[str, Any] = {} + + async def _get_client( redis_client: Any = None, redis_url: str | None = None, server: Any = None, ) -> Any: - """Build (or reuse) an async Redis client. Returns None if unavailable.""" + """Build (or reuse) an async Redis client. Returns None if unavailable. + + Real (URL-based) clients are cached module-level, keyed by URL, so + repeated gated calls (e.g. every ~30s polling tick) reuse one + redis.asyncio client/connection pool instead of constructing a new one + on every call. Test-injection paths (``redis_client=...`` or + ``server=...``) are deliberately NOT cached — each call still returns + the caller-provided client or a fresh ``FakeRedis`` wrapper, exactly as + before, so test isolation is unaffected. + """ if redis_client is not None: return redis_client if server is not None: @@ -159,9 +171,14 @@ async def _get_client( if url is None: _warn_no_redis_configured_once() return None + cached = _client_cache.get(url) + if cached is not None: + return cached import redis.asyncio as aioredis - return aioredis.from_url(url) + client = aioredis.from_url(url) + _client_cache[url] = client + return client async def set_component_due( diff --git a/tests/shared/test_notify_due.py b/tests/shared/test_notify_due.py index 0c469ae..26041c4 100644 --- a/tests/shared/test_notify_due.py +++ b/tests/shared/test_notify_due.py @@ -89,19 +89,6 @@ async def test_set_component_due_updates_existing_component_independently(): assert await get_due_user_ids(now + 10, server=server) == [] # neither passed yet -async def test_set_component_due_meeting_component_from_premium_contributes(): - """Premium contributes a 'meeting' component to the SAME due_queue — the - cron's single range query must see meeting-only-due users too.""" - from shared.notify_due import get_due_user_ids, set_component_due - - server = fakeredis.FakeServer() - now = time.time() - - await set_component_due("user-d", "meeting", now - 1, server=server) - - assert await get_due_user_ids(now, server=server) == ["user-d"] - - # --------------------------------------------------------------------------- # is_due # --------------------------------------------------------------------------- @@ -355,3 +342,69 @@ async def test_no_redis_configured_warning_logs_only_once_per_process(monkeypatc f"expected exactly one ERROR log across all 4 calls, got {len(error_records)}: " f"{[r.message for r in error_records]}" ) + + +# --------------------------------------------------------------------------- +# _get_client: module-level client caching keyed by URL (A9) +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def _reset_client_cache(): + """The URL->client cache is process-global state — reset it around every + test in this module so tests don't leak cached clients into each other.""" + import shared.notify_due as notify_due_module + + notify_due_module._client_cache.clear() + yield + notify_due_module._client_cache.clear() + + +async def test_get_client_reuses_same_client_for_same_url(): + from shared.notify_due import _get_client + + first = await _get_client(redis_url="redis://localhost:6379/0") + second = await _get_client(redis_url="redis://localhost:6379/0") + + assert first is second + + +async def test_get_client_different_urls_get_different_clients(): + from shared.notify_due import _get_client + + a = await _get_client(redis_url="redis://localhost:6379/0") + b = await _get_client(redis_url="redis://localhost:6379/1") + + assert a is not b + + +async def test_get_client_injected_redis_client_bypasses_cache(): + """An explicitly-injected redis_client (test/caller override) must be + returned as-is every time, never substituted with a cached instance.""" + from shared.notify_due import _get_client + + injected = object() + + result1 = await _get_client(redis_client=injected, redis_url="redis://localhost:6379/0") + result2 = await _get_client(redis_client=injected, redis_url="redis://localhost:6379/0") + + assert result1 is injected + assert result2 is injected + + +async def test_get_client_server_injection_still_returns_fresh_fakeredis_each_call(): + """The fakeredis `server=` test-injection path must be unaffected by the + URL cache — existing tests share state via the FakeServer, not via a + cached client object, and must keep working exactly as before.""" + from shared.notify_due import _get_client + + server = fakeredis.FakeServer() + + client1 = await _get_client(server=server) + client2 = await _get_client(server=server) + + # Not the same object (each call wraps a fresh FakeRedis), but both + # share the same underlying FakeServer so data written via one is + # visible via the other -- this is what existing tests rely on. + assert client1 is not client2 + await client1.set("k", "v") + assert await client2.get("k") == b"v" diff --git a/tether_mcp/server.py b/tether_mcp/server.py index 31758c9..4255cb1 100644 --- a/tether_mcp/server.py +++ b/tether_mcp/server.py @@ -58,7 +58,9 @@ def anchor_start(a: dict) -> datetime: return active -# ─── 9 Consolidated MCP Tools ─────────────────────────────────────────────── +# ─── MCP Tools (16 registered; consolidation toward a smaller intent-based +# surface is tracked separately — see cc-context-store/tether/docs/mcp/ +# mcp-consolidation-spec.md) ───────────────────────────────────────────────── @mcp.tool() async def upsert_tasks(tasks: list[dict]) -> list[dict]: diff --git a/tether_mcp/tools/read_context.py b/tether_mcp/tools/read_context.py index c7ede24..dc749b4 100644 --- a/tether_mcp/tools/read_context.py +++ b/tether_mcp/tools/read_context.py @@ -48,8 +48,8 @@ async def _build_node_response( _cascade_depth tracks levels descended from the read_context source node; when it reaches N, _add_children will not recurse further. """ - from db.pg_queries import get_node, get_children, get_sections, get_node_tasks - from db.pg_queries.node_memory import get_node_summary, log_node_read, get_node_tree_distance + from db.pg_queries import get_node, get_sections, get_node_tasks + from db.pg_queries.node_memory import get_node_summary, log_node_read from tether_mcp.write_modes import format_cat_n, line_count # Ensure we have full node dict (with section_types and children_count)