From 4a5485bc275f0dcb6f673c7069e23195b0daad6d Mon Sep 17 00:00:00 2001 From: Jason Lunder Date: Sat, 4 Jul 2026 11:02:09 -0700 Subject: [PATCH 1/7] A1: remove dead API-process meeting-event drain block _run_notification_check's call to drain_meeting_events was always a no-op in this process: the event queue it reads from is an in-process queue.Queue owned by tether_premium.bot.scheduling.events, populated by a WS-listener thread that only ever starts inside bot/message_handler.py's run_polling() (the BOT process). The API process (where _run_notification_check runs) cannot see that queue. Meeting-event draining happens exclusively via the bot's polling loop, unaffected by this change. Delete the dead block (import, _notify_meeting_send closure, meeting_send lambda, drain_meeting_events call) and update the surrounding docstrings (module header, _run_notification_check) to state this explicitly instead of implying meeting events are handled here. --- api/routes/internal.py | 40 ++++++++++++++-------------------------- 1 file changed, 14 insertions(+), 26 deletions(-) diff --git a/api/routes/internal.py b/api/routes/internal.py index 6286d08..73a3165 100644 --- a/api/routes/internal.py +++ b/api/routes/internal.py @@ -7,10 +7,17 @@ ``_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 @@ -199,11 +206,14 @@ async def _recompute_and_cache_due( 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") @@ -269,28 +279,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) From 5e5058868b2d4c52299c7ad0681fc76826b896db Mon Sep 17 00:00:00 2001 From: Jason Lunder Date: Sat, 4 Jul 2026 11:07:32 -0700 Subject: [PATCH 2/7] A2: collapse _recompute_and_cache_due (~35 -> ~15 lines) Drop the unused now_ts local (dead "reserved for future observability" placeholder that was never read) and the two empty else: pass branches (a None estimate already means "nothing to write" without needing an explicit no-op branch to say so). Also fixes a stale docstring claim that tether-premium's drain_meeting_events contributes a "meeting" component via this same mechanism -- confirmed there are zero writers of that component anywhere in the codebase today (see A7). Behavior unchanged: anchor/followup components are still written only when a non-None estimate is available. --- api/routes/internal.py | 27 +++++---------------------- 1 file changed, 5 insertions(+), 22 deletions(-) diff --git a/api/routes/internal.py b/api/routes/internal.py index 73a3165..ca2fc84 100644 --- a/api/routes/internal.py +++ b/api/routes/internal.py @@ -176,33 +176,16 @@ 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: From c4460df94f51d72a7a0c6723705b0029677441e4 Mon Sep 17 00:00:00 2001 From: Jason Lunder Date: Sat, 4 Jul 2026 11:08:45 -0700 Subject: [PATCH 3/7] A3: delete redundant outer try/except around _check_anchor_transitions _check_anchor_transitions already wraps its entire body in a catch-all except Exception (logs at ERROR, returns None) -- it can never raise, so the outer try/except at its call site in _run_notification_check was dead code that could never trigger. Removed; the inner catch remains the sole (and sufficient) error boundary. The sibling check_followups() call keeps its own try/except since that function has no internal catch-all. --- api/routes/internal.py | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/api/routes/internal.py b/api/routes/internal.py index ca2fc84..a90e1aa 100644 --- a/api/routes/internal.py +++ b/api/routes/internal.py @@ -236,14 +236,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: From ed52a0a7d46474fa69d44533980c676fee5dc76b Mon Sep 17 00:00:00 2001 From: Jason Lunder Date: Sat, 4 Jul 2026 11:10:14 -0700 Subject: [PATCH 4/7] A5: remove dead get_children/get_node_tree_distance imports in _build_node_response Both were imported at the top of _build_node_response() but never called directly in its body -- all actual usage happens in _add_children(), which already has its own local import of the same names. No behavior change. --- tether_mcp/tools/read_context.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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) From dc114262ce77ccad02d0a7c1a1aab58a0a25b4e5 Mon Sep 17 00:00:00 2001 From: Jason Lunder Date: Sat, 4 Jul 2026 11:13:55 -0700 Subject: [PATCH 5/7] A6: fix stale comments (MCP tool count, cron wording) - tether_mcp/server.py: header comment claimed "9 Consolidated MCP Tools" but 16 tools are registered today (verified: `grep -c '^@mcp.tool()'`). Updated to state the actual count and point at the consolidation spec as a separate tracked effort, rather than asserting a target as fact. - api/routes/internal.py: module docstring and /notifications/check's docstring both asserted "Called by Fly.io cron [machines/every minute]" -- unverified (no Fly scheduled-machine config exists anywhere in this repo; the actual trigger mechanism/cadence is out-of-band deployment config). Reworded to describe the endpoint's actual contract (internal, token-gated, notify_due-gated so safe at any cadence) without asserting a specific unverified scheduling mechanism. --- api/routes/internal.py | 7 +++++-- tether_mcp/server.py | 4 +++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/api/routes/internal.py b/api/routes/internal.py index a90e1aa..e72a4ea 100644 --- a/api/routes/internal.py +++ b/api/routes/internal.py @@ -1,6 +1,8 @@ """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 ------------------------- @@ -52,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) 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]: From d1854159ff10b241fd3b144910a72f01238eed95 Mon Sep 17 00:00:00 2001 From: Jason Lunder Date: Sat, 4 Jul 2026 11:18:24 -0700 Subject: [PATCH 6/7] A7: remove "meeting" from notify_due Component Literal Verified zero writers anywhere in tether or tether-premium (grepped both checkouts for the "meeting" component string) -- the reservation was never used and, per A1, the code path that would have used it (API-process meeting drain) was itself dead. Removed the Literal value, rewrote the docstring paragraph to state plainly that meeting events don't fit this module's model rather than describing a reservation nothing consumes, and deleted the test that only existed to exercise that reservation. --- shared/notify_due.py | 18 +++++++++--------- tests/shared/test_notify_due.py | 13 ------------- 2 files changed, 9 insertions(+), 22 deletions(-) diff --git a/shared/notify_due.py b/shared/notify_due.py index 190b4a1..a5029f0 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: diff --git a/tests/shared/test_notify_due.py b/tests/shared/test_notify_due.py index 0c469ae..6366685 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 # --------------------------------------------------------------------------- From ff3fd9e7ac5bb6750bc3c1c9f3cbc441ef3918f4 Mon Sep 17 00:00:00 2001 From: Jason Lunder Date: Sat, 4 Jul 2026 11:23:25 -0700 Subject: [PATCH 7/7] A9: cache Redis client module-level, keyed by URL _get_client previously constructed a brand-new redis.asyncio client (and its underlying connection pool) on every single gated call -- e.g. every ~30s polling tick in bot/message_handler.py's run_polling, or every cron invocation of _run_notification_check -- rather than reusing one client per process. Added a module-level dict cache keyed by REDIS_URL so real (URL-based) clients are constructed once and reused. Test-injection paths are deliberately untouched: an explicitly-passed redis_client is still returned as-is every call (never substituted with a cached instance), and the fakeredis `server=` path still returns a fresh FakeRedis wrapper per call (sharing state via the FakeServer, not via a cached client object) -- exactly the behavior every existing test already depends on. 4 new tests cover the caching behavior and confirm both test-injection paths remain uncached. --- shared/notify_due.py | 21 ++++++++++- tests/shared/test_notify_due.py | 66 +++++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+), 2 deletions(-) diff --git a/shared/notify_due.py b/shared/notify_due.py index a5029f0..94fff9f 100644 --- a/shared/notify_due.py +++ b/shared/notify_due.py @@ -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 6366685..26041c4 100644 --- a/tests/shared/test_notify_due.py +++ b/tests/shared/test_notify_due.py @@ -342,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"