From 85ba83f08d0574ba07cb05b8df1d2cdf00de0b2d Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Mon, 26 Oct 2026 05:03:56 -0700 Subject: [PATCH 1/4] feat(gateway): optional PostHog signup, identify and activation tracking Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- CHANGELOG.md | 3 + gateway/.env.example | 9 + gateway/app/core/product_analytics.py | 117 +++++++++++++ gateway/app/routers/billing.py | 7 +- gateway/app/routers/demo_keys.py | 2 + gateway/app/routers/dev_keys.py | 2 + gateway/app/services/api_key_service.py | 78 ++++++++- gateway/docs/README.md | 2 + gateway/docs/product-analytics.md | 45 +++++ gateway/tests/test_product_analytics.py | 211 ++++++++++++++++++++++++ 10 files changed, 469 insertions(+), 7 deletions(-) create mode 100644 gateway/app/core/product_analytics.py create mode 100644 gateway/docs/product-analytics.md create mode 100644 gateway/tests/test_product_analytics.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d8688fbe..4815affa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,9 @@ This repository starts with a fresh history. Releases are listed here once tagge ## [Unreleased] +- Optional PostHog product tracking (`POSTHOG_API_KEY`, `POSTHOG_HOST`; off when + unset): identify with email and `signup` on account creation, `activation` once + per account on first key use. Fail-safe and non-blocking. - Upstream MCP: optional multi-tool discovery and exact-name allowlist with stable public IDs, per-tool permit scope, and unchanged per-call credits. - Initial import of the gateway (Fair Source, FSL-1.1-ALv2), the Python SDK and diff --git a/gateway/.env.example b/gateway/.env.example index 3ed81dd6..8a3fe833 100644 --- a/gateway/.env.example +++ b/gateway/.env.example @@ -17,6 +17,15 @@ PORT=8000 # PUBLIC_URL=https://api.thisisatest.tech PUBLIC_URL= +# --- Product analytics (PostHog, optional) --- +# Read from the process environment only. Leave POSTHOG_API_KEY blank (the +# default) and tracking is a no-op. Never commit a real key. +# Events: signup + identify (email) on account creation, activation on the +# account's first key use. See docs/product-analytics.md. +POSTHOG_API_KEY= +# POSTHOG_HOST=https://us.i.posthog.com +POSTHOG_HOST= + # Public OpenAPI contact metadata. Leave blank until these are real, monitored # operator details; placeholder contact information must never be deployed. PUBLIC_CONTACT_NAME= diff --git a/gateway/app/core/product_analytics.py b/gateway/app/core/product_analytics.py new file mode 100644 index 00000000..ad1ccf7b --- /dev/null +++ b/gateway/app/core/product_analytics.py @@ -0,0 +1,117 @@ +"""Fail-safe, non-blocking PostHog product tracking. + +Tracking is off unless ``POSTHOG_API_KEY`` is set in the environment. When it +is unset every function here is a no-op: no client, no network, no task. + +Events go to PostHog's public ``/capture/`` endpoint over httpx (already a +gateway dependency), so no PostHog SDK is installed. Each send is scheduled as +a background task with a short timeout, and every exception is caught and +logged at debug level without the API key, so tracking can never fail or slow +a request. + +The distinct id is the account's root (sponsor) wallet id. Only +:func:`identify` sends personal data: the ``email`` person property. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +from typing import Any + +import httpx + +logger = logging.getLogger(__name__) + +DEFAULT_HOST = "https://us.i.posthog.com" +SEND_TIMEOUT_SECONDS = 2.0 + +EVENT_SIGNUP = "signup" +EVENT_ACTIVATION = "activation" + +_pending: set[asyncio.Task[None]] = set() + + +def _config() -> tuple[str, str] | None: + api_key = os.environ.get("POSTHOG_API_KEY", "").strip() + if not api_key: + return None + host = os.environ.get("POSTHOG_HOST", "").strip() or DEFAULT_HOST + return api_key, host.rstrip("/") + + +def is_enabled() -> bool: + return _config() is not None + + +async def _post(host: str, payload: dict[str, Any]) -> None: + async with httpx.AsyncClient(timeout=SEND_TIMEOUT_SECONDS) as client: + response = await client.post(f"{host}/capture/", json=payload) + response.raise_for_status() + + +async def _send(host: str, payload: dict[str, Any]) -> None: + try: + await _post(host, payload) + except Exception as exc: # noqa: BLE001 - tracking must never raise + logger.debug( + "product_analytics_send_failed event=%s error=%s", + payload.get("event"), + type(exc).__name__, + ) + + +def _enqueue(event: str, distinct_id: str, properties: dict[str, Any]) -> None: + try: + config = _config() + if config is None or not distinct_id: + return + api_key, host = config + payload = { + "api_key": api_key, + "event": event, + "distinct_id": distinct_id, + "properties": {**properties, "$lib": "agent-middleware-gateway"}, + } + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return # no event loop (sync caller): drop rather than block + task = loop.create_task(_send(host, payload)) + _pending.add(task) + task.add_done_callback(_pending.discard) + except Exception as exc: # noqa: BLE001 - tracking must never raise + logger.debug( + "product_analytics_enqueue_failed event=%s error=%s", + event, + type(exc).__name__, + ) + + +def identify(distinct_id: str, *, email: str | None) -> None: + """Attach the email person property to the account.""" + props: dict[str, Any] = {} + if email: + props["$set"] = {"email": email} + _enqueue("$identify", distinct_id, props) + + +def capture(event: str, distinct_id: str, **properties: Any) -> None: + _enqueue(event, distinct_id, properties) + + +def track_signup(distinct_id: str, *, method: str, email: str | None = None) -> None: + """identify (with email when known) then a ``signup`` event.""" + identify(distinct_id, email=email) + capture(EVENT_SIGNUP, distinct_id, method=method) + + +def track_activation(distinct_id: str, *, key_id: str, wallet_id: str) -> None: + capture(EVENT_ACTIVATION, distinct_id, key_id=key_id, wallet_id=wallet_id) + + +async def drain_pending() -> None: + """Await in-flight sends (tests and graceful shutdown).""" + if _pending: + await asyncio.gather(*list(_pending), return_exceptions=True) diff --git a/gateway/app/routers/billing.py b/gateway/app/routers/billing.py index aff54833..d14eea8c 100644 --- a/gateway/app/routers/billing.py +++ b/gateway/app/routers/billing.py @@ -18,6 +18,7 @@ from fastapi.responses import JSONResponse from pydantic import BaseModel, Field +from ..core import product_analytics from ..core.auth import AuthContext, get_auth_context, verify_api_key from ..core.config import get_settings from ..core.wallet_resolution import resolve_wallet_id @@ -280,7 +281,7 @@ async def create_sponsor_wallet( if request.initial_credits else Decimal("0") ) - return await money.create_sponsor_wallet( + wallet = await money.create_sponsor_wallet( sponsor_name=request.sponsor_name, email=request.email, initial_credits=initial, @@ -288,6 +289,10 @@ async def create_sponsor_wallet( metadata=request.metadata, require_kyc=request.require_kyc, ) + product_analytics.track_signup( + wallet.wallet_id, method="sponsor_wallet", email=request.email + ) + return wallet @router.post( diff --git a/gateway/app/routers/demo_keys.py b/gateway/app/routers/demo_keys.py index 2fabd8e2..a4e81fba 100644 --- a/gateway/app/routers/demo_keys.py +++ b/gateway/app/routers/demo_keys.py @@ -10,6 +10,7 @@ from pydantic import BaseModel, Field, field_validator from ..core.auth import AuthContext, get_auth_context +from ..core import product_analytics from ..core.config import get_settings from ..core.rate_limiter import _client_id from ..services.demo_tenant import ( @@ -225,6 +226,7 @@ async def issue_demo_key( key_id=minted["key_id"], wallet_id=minted["wallet_id"], ) + product_analytics.track_signup(minted["sponsor_wallet_id"], method="demo_key") base_url = settings.PUBLIC_URL.rstrip("/") if settings.PUBLIC_URL else "" return DemoKeyIssueResponse( api_key=minted["api_key"], diff --git a/gateway/app/routers/dev_keys.py b/gateway/app/routers/dev_keys.py index cc93a87c..2c266018 100644 --- a/gateway/app/routers/dev_keys.py +++ b/gateway/app/routers/dev_keys.py @@ -39,6 +39,7 @@ from fastapi import APIRouter, HTTPException, Request, status from pydantic import BaseModel, Field +from ..core import product_analytics from ..core.browser_origin import validate_browser_origin from ..core.config import get_settings from ..core.trust_mode import is_production_like_environment @@ -171,6 +172,7 @@ async def self_provision_dev_key( key_name=request.key_name, ) + product_analytics.track_signup(sponsor.wallet_id, method="dev_self_provision") return SelfProvisionResponse( sponsor_wallet_id=sponsor.wallet_id, wallet_id=agent.wallet_id, diff --git a/gateway/app/services/api_key_service.py b/gateway/app/services/api_key_service.py index b15bdaab..dd86a6aa 100644 --- a/gateway/app/services/api_key_service.py +++ b/gateway/app/services/api_key_service.py @@ -24,6 +24,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel import col +from ..core import product_analytics from ..core.time import to_naive_utc, utc_now from ..db.database import get_session_factory from ..db.models import ( @@ -484,6 +485,20 @@ async def validate_key( if expires_at and expires_at < now: return None + # A key read as never-used leaves first_used_at to the guarded + # claim below, so exactly one request records the first use (and + # fires the activation event) even when first uses race. + unused_at_read = key.first_used_at is None + first_use_values: dict[str, Any] = ( + {} + if unused_at_read + else { + "first_used_at": func.coalesce( + APIKeyModel.first_used_at, to_naive_utc(now) + ) + } + ) + if key.max_uses is not None: # Guarded increment: the WHERE clause makes concurrent # requests race for the remaining budget instead of all @@ -507,9 +522,7 @@ async def validate_key( .values( use_count=APIKeyModel.use_count + 1, last_used_at=to_naive_utc(now), - first_used_at=func.coalesce( - APIKeyModel.first_used_at, to_naive_utc(now) - ), + **first_use_values, ) .execution_options(synchronize_session=False) ) @@ -529,9 +542,7 @@ async def validate_key( ) .values( last_used_at=to_naive_utc(now), - first_used_at=func.coalesce( - APIKeyModel.first_used_at, to_naive_utc(now) - ), + **first_use_values, ) .execution_options(synchronize_session=False) ) @@ -539,8 +550,63 @@ async def validate_key( if (cast(Any, used).rowcount or 0) == 0: return None + if unused_at_read: + claimed = await session.execute( + update(APIKeyModel) + .where( + col(APIKeyModel.key_id) == key.key_id, + col(APIKeyModel.first_used_at).is_(None), + ) + .values(first_used_at=to_naive_utc(now)) + .execution_options(synchronize_session=False) + ) + await session.commit() + if (cast(Any, claimed).rowcount or 0) == 1: + await self._track_first_key_use(session, key) + return key + async def _track_first_key_use( + self, session: AsyncSession, key: APIKeyModel + ) -> None: + """Fire ``activation`` once per account on its first key use. + + The account is the root (sponsor) wallet. The event fires only when + no other key on the root wallet or its direct child wallets has been + used. Never raises: tracking must not affect authentication. + """ + if not product_analytics.is_enabled(): + return + try: + wallet = await session.get(WalletModel, key.wallet_id) + root_id = key.wallet_id + if wallet is not None and wallet.parent_wallet_id: + root_id = wallet.parent_wallet_id + scope = select(col(WalletModel.wallet_id)).where( + or_( + col(WalletModel.wallet_id) == root_id, + col(WalletModel.parent_wallet_id) == root_id, + ) + ) + prior = await session.execute( + select(func.count()) + .select_from(APIKeyModel) + .where( + col(APIKeyModel.key_id) != key.key_id, + col(APIKeyModel.first_used_at).is_not(None), + col(APIKeyModel.wallet_id).in_(scope), + ) + ) + if (prior.scalar_one() or 0) > 0: + return + product_analytics.track_activation( + root_id, key_id=key.key_id, wallet_id=key.wallet_id + ) + except Exception as exc: # noqa: BLE001 - tracking must never raise + logger.debug( + "product_analytics_activation_failed error=%s", type(exc).__name__ + ) + async def rotate_key( self, wallet_id: str, diff --git a/gateway/docs/README.md b/gateway/docs/README.md index 5ddb9001..3446eff9 100644 --- a/gateway/docs/README.md +++ b/gateway/docs/README.md @@ -49,6 +49,8 @@ use them as product onboarding or a public capability list: - [`agent-recipes.md`](agent-recipes.md) and [`human-onboarding.md`](human-onboarding.md) are legacy proof-surface material, not the supported one-tool pilot. +- [Product analytics](product-analytics.md) documents the optional PostHog + signup, identify and activation events (`POSTHOG_API_KEY`, `POSTHOG_HOST`). - [Pods](pods.md) are a dormant trust feature, not part of the supported pilot. - `server.json` at the repository root is the official MCP registry manifest. It advertises the standard MCP endpoint diff --git a/gateway/docs/product-analytics.md b/gateway/docs/product-analytics.md new file mode 100644 index 00000000..78eb9e8c --- /dev/null +++ b/gateway/docs/product-analytics.md @@ -0,0 +1,45 @@ +# Product analytics (PostHog) + +The gateway can send three product events to PostHog. It is off unless +`POSTHOG_API_KEY` is set. + +| Env var | Default | Meaning | +| --- | --- | --- | +| `POSTHOG_API_KEY` | empty (off) | PostHog project API key. Blank or unset: every tracking call is a no-op. | +| `POSTHOG_HOST` | `https://us.i.posthog.com` | PostHog ingestion host, e.g. `https://eu.i.posthog.com` or a self-hosted URL. | + +Both are read from the process environment at send time (not from `Settings`, +so they never appear in a settings dump). Never commit a real key. + +## Events + +The distinct id is the account's root (sponsor) wallet id. + +| Event | Where | Properties | +| --- | --- | --- | +| `$identify` | Every signup, just before `signup` | `$set.email` when the signup carries an email (sponsor wallets only) | +| `signup` | `POST /v1/billing/wallets/sponsor` (`method=sponsor_wallet`), `POST /v1/demo/keys` (`method=demo_key`), `POST /v1/dev-keys/self-provision` (`method=dev_self_provision`) | `method` | +| `activation` | First successful API key authentication for the account (`APIKeyService.validate_key`) | `key_id`, `wallet_id` | + +Demo and dev self-provision signups are anonymous: no email is sent (dev +wallets only carry a synthetic `@dev.local` address). + +`activation` fires once per account. The key's first use is claimed with a +guarded `first_used_at IS NULL` update, so concurrent first uses produce one +event, and it is skipped when another key on the root wallet or its direct +child wallets was already used. Derived (JWT) authority recorded through +`consume_derived_key_use` is not an activation trigger. + +## Failure behaviour + +Sends go to `/capture/` over httpx as background tasks with a +2 s timeout. Every error is caught and logged at debug level with the +exception type only (never the key), so tracking cannot fail or slow a +request. Code lives in `app/core/product_analytics.py`; tests in +`tests/test_product_analytics.py` mock the HTTP send. + +## Privacy + +Only `$identify` sends personal data (the sponsor email). Enabling tracking +in a deployment is an operator decision and should be reflected in that +deployment's privacy notice. diff --git a/gateway/tests/test_product_analytics.py b/gateway/tests/test_product_analytics.py new file mode 100644 index 00000000..acabd513 --- /dev/null +++ b/gateway/tests/test_product_analytics.py @@ -0,0 +1,211 @@ +"""PostHog product tracking: identify/signup/activation, fail-safe, off by default. + +The PostHog HTTP send (``product_analytics._post``) is mocked throughout; no +test touches the network. +""" + +import asyncio +import logging + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.core import product_analytics +from app.core.config import get_settings +from app.main import app +from app.services.api_key_service import get_api_key_service + +FAKE_KEY = "phc_test_not_a_real_key" + + +@pytest.fixture +def sent(monkeypatch): + calls: list[tuple[str, dict]] = [] + + async def fake_post(host, payload): + calls.append((host, payload)) + + monkeypatch.setattr(product_analytics, "_post", fake_post) + return calls + + +@pytest.fixture +def enabled(monkeypatch): + monkeypatch.setenv("POSTHOG_API_KEY", FAKE_KEY) + monkeypatch.setenv("POSTHOG_HOST", "https://posthog.example/") + + +@pytest.fixture +async def client(clean_database): + transport = ASGITransport(app=app) + async with AsyncClient( + transport=transport, + base_url="http://test", + headers={"X-API-Key": "test-key"}, + ) as c: + yield c + + +@pytest.fixture +def self_provision(monkeypatch): + settings = get_settings() + monkeypatch.setattr(settings, "ENVIRONMENT", "local") + monkeypatch.setattr(settings, "ENABLE_DEV_KEY_SELF_PROVISION", True) + + +def _events(calls, distinct_id=None): + return [ + p["event"] + for _, p in calls + if distinct_id is None or p["distinct_id"] == distinct_id + ] + + +async def test_noop_without_api_key(monkeypatch, sent): + monkeypatch.delenv("POSTHOG_API_KEY", raising=False) + assert product_analytics.is_enabled() is False + product_analytics.track_signup("w1", method="x", email="a@b.dev") + product_analytics.track_activation("w1", key_id="k", wallet_id="w1") + await product_analytics.drain_pending() + assert sent == [] + assert not product_analytics._pending + + +async def test_blank_api_key_is_noop(monkeypatch, sent): + monkeypatch.setenv("POSTHOG_API_KEY", " ") + product_analytics.capture("signup", "w1") + await product_analytics.drain_pending() + assert sent == [] + + +async def test_identify_carries_email_and_signup_follows(enabled, sent): + product_analytics.track_signup("w1", method="sponsor_wallet", email="a@b.dev") + await product_analytics.drain_pending() + assert [p["event"] for _, p in sent] == ["$identify", "signup"] + host, ident = sent[0] + assert host == "https://posthog.example" + assert ident["distinct_id"] == "w1" + assert ident["api_key"] == FAKE_KEY + assert ident["properties"]["$set"] == {"email": "a@b.dev"} + assert sent[1][1]["properties"]["method"] == "sponsor_wallet" + + +async def test_default_host(monkeypatch, sent): + monkeypatch.setenv("POSTHOG_API_KEY", FAKE_KEY) + monkeypatch.delenv("POSTHOG_HOST", raising=False) + product_analytics.capture("signup", "w1") + await product_analytics.drain_pending() + assert sent[0][0] == product_analytics.DEFAULT_HOST + + +async def test_send_failure_is_swallowed_and_never_logs_key( + enabled, monkeypatch, caplog +): + async def boom(host, payload): + raise RuntimeError(f"failed with {payload['api_key']}") + + monkeypatch.setattr(product_analytics, "_post", boom) + caplog.set_level(logging.DEBUG, logger="app.core.product_analytics") + product_analytics.track_signup("w1", method="x", email="a@b.dev") + await product_analytics.drain_pending() + assert "product_analytics_send_failed" in caplog.text + assert FAKE_KEY not in caplog.text + + +async def test_slow_send_does_not_block_caller(enabled, monkeypatch): + gate = asyncio.Event() + + async def slow(host, payload): + await gate.wait() + + monkeypatch.setattr(product_analytics, "_post", slow) + product_analytics.capture("signup", "w1") # returns immediately + assert len(product_analytics._pending) == 1 + gate.set() + await product_analytics.drain_pending() + + +def test_no_running_loop_is_noop(enabled, sent): + product_analytics.capture("signup", "w1") + assert sent == [] + + +async def test_sponsor_wallet_signup_identifies_with_email(client, enabled, sent): + resp = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Tracking Sponsor", + "email": "tracking-sponsor@b2a.dev", + "initial_credits": 0, + }, + ) + assert resp.status_code == 201, resp.text + wallet_id = resp.json()["wallet_id"] + await product_analytics.drain_pending() + assert _events(sent, wallet_id) == ["$identify", "signup"] + assert sent[0][1]["properties"]["$set"]["email"] == "tracking-sponsor@b2a.dev" + + +async def test_request_succeeds_when_tracking_raises(client, enabled, monkeypatch): + def boom(*a, **k): + raise RuntimeError("tracking exploded") + + monkeypatch.setattr(product_analytics, "_config", boom) + resp = await client.post( + "/v1/billing/wallets/sponsor", + json={"sponsor_name": "S", "email": "s@b2a.dev", "initial_credits": 0}, + ) + assert resp.status_code == 201, resp.text + + +async def test_activation_fires_once_per_account(client, enabled, sent, self_provision): + resp = await client.post("/v1/dev-keys/self-provision", json={}) + assert resp.status_code == 201, resp.text + body = resp.json() + sponsor = body["sponsor_wallet_id"] + await product_analytics.drain_pending() + assert _events(sent, sponsor) == ["$identify", "signup"] + assert sent[1][1]["properties"]["method"] == "dev_self_provision" + + service = get_api_key_service() + # Non-consuming preflight is not a use. + assert await service.validate_key(body["api_key"], consume=False) + assert await service.validate_key(body["api_key"]) + assert await service.validate_key(body["api_key"]) + await product_analytics.drain_pending() + activations = [p for _, p in sent if p["event"] == "activation"] + assert len(activations) == 1 + assert activations[0]["distinct_id"] == sponsor + assert activations[0]["properties"]["key_id"] == body["key_id"] + assert activations[0]["properties"]["wallet_id"] == body["wallet_id"] + + # A second key on the same account: its first use is not an activation. + second = await service.create_key(wallet_id=body["wallet_id"], key_name="2nd") + assert await service.validate_key(second["api_key"]) + await product_analytics.drain_pending() + assert len([p for _, p in sent if p["event"] == "activation"]) == 1 + + +async def test_concurrent_first_uses_fire_one_activation( + client, enabled, sent, self_provision +): + body = (await client.post("/v1/dev-keys/self-provision", json={})).json() + service = get_api_key_service() + results = await asyncio.gather( + *(service.validate_key(body["api_key"]) for _ in range(5)) + ) + assert all(results) + await product_analytics.drain_pending() + assert len([p for _, p in sent if p["event"] == "activation"]) == 1 + + +async def test_first_use_recorded_when_tracking_disabled( + client, monkeypatch, sent, self_provision +): + monkeypatch.delenv("POSTHOG_API_KEY", raising=False) + body = (await client.post("/v1/dev-keys/self-provision", json={})).json() + service = get_api_key_service() + assert await service.validate_key(body["api_key"]) + key = await service.validate_key(body["api_key"], consume=False) + assert key.first_used_at is not None + assert sent == [] From 19e9500c253f42d7fd21fe0391dd92e22c7c4383 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Mon, 26 Oct 2026 05:28:01 -0700 Subject: [PATCH 2/4] fix(analytics): Settings-backed PostHog config, atomic per-account activation - Declare POSTHOG_API_KEY (SecretStr) and POSTHOG_HOST on Settings so .env entries load; product_analytics reads env, then Settings. - Claim activation atomically on wallets.product_activated_at (new migration 054 with backfill) via a conditional UPDATE. Walks parent_wallet_id to the root, bounded and cycle-safe. - Restore the single usage update and commit for first_used_at; drop the split claim. - Cap in-flight sends, reuse one httpx client, drain and close on shutdown. - Tests: two-key concurrency, claim failure, root walk, Settings/.env, cap, shutdown. Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- gateway/.env.example | 3 +- gateway/app/core/config.py | 4 + gateway/app/core/product_analytics.py | 50 ++++++- gateway/app/db/models.py | 5 + gateway/app/main.py | 4 + gateway/app/services/api_key_service.py | 110 ++++++++------- gateway/docs/product-analytics.md | 21 ++- gateway/docs/schema-054-rollout.md | 25 ++++ .../versions/054_wallet_product_activation.py | 53 +++++++ gateway/tests/test_product_analytics.py | 130 ++++++++++++++++++ 10 files changed, 336 insertions(+), 69 deletions(-) create mode 100644 gateway/docs/schema-054-rollout.md create mode 100644 gateway/migrations/versions/054_wallet_product_activation.py diff --git a/gateway/.env.example b/gateway/.env.example index 8a3fe833..a1715f96 100644 --- a/gateway/.env.example +++ b/gateway/.env.example @@ -18,7 +18,8 @@ PORT=8000 PUBLIC_URL= # --- Product analytics (PostHog, optional) --- -# Read from the process environment only. Leave POSTHOG_API_KEY blank (the +# Read from the process environment, then this file (Settings). Leave +# POSTHOG_API_KEY blank (the # default) and tracking is a no-op. Never commit a real key. # Events: signup + identify (email) on account creation, activation on the # account's first key use. See docs/product-analytics.md. diff --git a/gateway/app/core/config.py b/gateway/app/core/config.py index d4bb25a2..4e6f354f 100644 --- a/gateway/app/core/config.py +++ b/gateway/app/core/config.py @@ -248,6 +248,10 @@ def demo_allowed_tools_list(self) -> list[str]: JEV_RISK_GUARD_TIMEOUT_SECONDS: float = 1.5 JEV_RISK_GUARD_TIERS: str = "medium,high" TYPESAFE_API_KEY: SecretStr = SecretStr("") + # Optional PostHog product tracking (docs/product-analytics.md). Blank + # key = tracking off. Declared here so .env entries load. + POSTHOG_API_KEY: SecretStr = SecretStr("") + POSTHOG_HOST: str = "" TYPESAFE_BASE_URL: str = "https://api.typesafe.ai" # Default repeat detection window in seconds. Duplicates outside this window # are not detected. diff --git a/gateway/app/core/product_analytics.py b/gateway/app/core/product_analytics.py index ad1ccf7b..64a1884b 100644 --- a/gateway/app/core/product_analytics.py +++ b/gateway/app/core/product_analytics.py @@ -1,6 +1,7 @@ """Fail-safe, non-blocking PostHog product tracking. -Tracking is off unless ``POSTHOG_API_KEY`` is set in the environment. When it +Tracking is off unless ``POSTHOG_API_KEY`` is set (process env or ``.env`` via +Settings). When it is unset every function here is a no-op: no client, no network, no task. Events go to PostHog's public ``/capture/`` endpoint over httpx (already a @@ -30,25 +31,46 @@ EVENT_SIGNUP = "signup" EVENT_ACTIVATION = "activation" +# In-flight sends are capped; beyond this, new events are dropped so a slow or +# down PostHog can never pile up unbounded tasks. +MAX_PENDING = 256 + _pending: set[asyncio.Task[None]] = set() +_client: httpx.AsyncClient | None = None def _config() -> tuple[str, str] | None: + """(api_key, host) from process env first, then Settings (``.env``).""" + from app.core.config import get_settings + + settings = get_settings() api_key = os.environ.get("POSTHOG_API_KEY", "").strip() + if not api_key: + api_key = settings.POSTHOG_API_KEY.get_secret_value().strip() if not api_key: return None - host = os.environ.get("POSTHOG_HOST", "").strip() or DEFAULT_HOST + host = ( + os.environ.get("POSTHOG_HOST", "").strip() + or settings.POSTHOG_HOST.strip() + or DEFAULT_HOST + ) return api_key, host.rstrip("/") +def _get_client() -> httpx.AsyncClient: + global _client + if _client is None or _client.is_closed: + _client = httpx.AsyncClient(timeout=SEND_TIMEOUT_SECONDS) + return _client + + def is_enabled() -> bool: return _config() is not None async def _post(host: str, payload: dict[str, Any]) -> None: - async with httpx.AsyncClient(timeout=SEND_TIMEOUT_SECONDS) as client: - response = await client.post(f"{host}/capture/", json=payload) - response.raise_for_status() + response = await _get_client().post(f"{host}/capture/", json=payload) + response.raise_for_status() async def _send(host: str, payload: dict[str, Any]) -> None: @@ -67,6 +89,9 @@ def _enqueue(event: str, distinct_id: str, properties: dict[str, Any]) -> None: config = _config() if config is None or not distinct_id: return + if len(_pending) >= MAX_PENDING: + logger.debug("product_analytics_dropped event=%s reason=backlog", event) + return api_key, host = config payload = { "api_key": api_key, @@ -115,3 +140,18 @@ async def drain_pending() -> None: """Await in-flight sends (tests and graceful shutdown).""" if _pending: await asyncio.gather(*list(_pending), return_exceptions=True) + + +async def shutdown() -> None: + """Drain in-flight sends and close the shared client (app shutdown).""" + global _client + try: + await asyncio.wait_for(drain_pending(), timeout=SEND_TIMEOUT_SECONDS + 1) + except Exception as exc: # noqa: BLE001 - shutdown must not raise + logger.debug("product_analytics_drain_failed error=%s", type(exc).__name__) + client, _client = _client, None + if client is not None: + try: + await client.aclose() + except Exception as exc: # noqa: BLE001 + logger.debug("product_analytics_close_failed error=%s", type(exc).__name__) diff --git a/gateway/app/db/models.py b/gateway/app/db/models.py index 474be7e4..437f4295 100644 --- a/gateway/app/db/models.py +++ b/gateway/app/db/models.py @@ -53,6 +53,11 @@ class WalletModel(SQLModel, table=True): daily_limit: Optional[Decimal] = Field(default=None, decimal_places=8) daily_spent: Decimal = Field(default=Decimal("0"), decimal_places=8) daily_reset_at: Optional[datetime] = Field(sa_type=NaiveUTCDateTime, default=None) + # Account-level product activation claim (root wallets only), set once by + # a conditional UPDATE on the first key use; see 054 migration. + product_activated_at: Optional[datetime] = Field( + sa_type=NaiveUTCDateTime, default=None + ) # Velocity monitoring (spend per hour for anomaly detection) hourly_limit: Optional[Decimal] = Field(default=None, decimal_places=8) diff --git a/gateway/app/main.py b/gateway/app/main.py index 5282ee09..54738b11 100644 --- a/gateway/app/main.py +++ b/gateway/app/main.py @@ -496,6 +496,10 @@ async def periodic_cleanup(): logger.info("app_shutdown", phase="starting") shutdown_start = time.monotonic() + from .core import product_analytics + + await product_analytics.shutdown() + if cleanup_task: cleanup_task.cancel() try: diff --git a/gateway/app/services/api_key_service.py b/gateway/app/services/api_key_service.py index dd86a6aa..7120c123 100644 --- a/gateway/app/services/api_key_service.py +++ b/gateway/app/services/api_key_service.py @@ -40,6 +40,10 @@ logger = logging.getLogger(__name__) +# Wallets nest via parent_wallet_id (sponsor -> agent -> child agent); the +# root walk is capped so a corrupt cycle cannot loop forever. +_MAX_WALLET_DEPTH = 16 + API_KEY_LENGTH = 32 API_KEY_PREFIX_LENGTH = 8 @@ -485,19 +489,14 @@ async def validate_key( if expires_at and expires_at < now: return None - # A key read as never-used leaves first_used_at to the guarded - # claim below, so exactly one request records the first use (and - # fires the activation event) even when first uses race. + # Detect a first use from the pre-read; first_used_at itself is + # still set by the single guarded update below (one commit). unused_at_read = key.first_used_at is None - first_use_values: dict[str, Any] = ( - {} - if unused_at_read - else { - "first_used_at": func.coalesce( - APIKeyModel.first_used_at, to_naive_utc(now) - ) - } - ) + first_use_values: dict[str, Any] = { + "first_used_at": func.coalesce( + APIKeyModel.first_used_at, to_naive_utc(now) + ) + } if key.max_uses is not None: # Guarded increment: the WHERE clause makes concurrent @@ -551,61 +550,60 @@ async def validate_key( return None if unused_at_read: - claimed = await session.execute( - update(APIKeyModel) - .where( - col(APIKeyModel.key_id) == key.key_id, - col(APIKeyModel.first_used_at).is_(None), - ) - .values(first_used_at=to_naive_utc(now)) - .execution_options(synchronize_session=False) - ) - await session.commit() - if (cast(Any, claimed).rowcount or 0) == 1: - await self._track_first_key_use(session, key) + await self._claim_account_activation(session, key) return key - async def _track_first_key_use( + async def _claim_account_activation( self, session: AsyncSession, key: APIKeyModel ) -> None: - """Fire ``activation`` once per account on its first key use. - - The account is the root (sponsor) wallet. The event fires only when - no other key on the root wallet or its direct child wallets has been - used. Never raises: tracking must not affect authentication. + """Claim the account's activation once and fire ``activation``. + + The account is the root wallet, found by walking ``parent_wallet_id``. + The claim is a conditional UPDATE on + ``wallets.product_activated_at IS NULL``, so concurrent first uses of + the same key or of different keys on one account win it exactly once. + This runs only on a key's first use. Never raises: activation must not + affect authentication. """ - if not product_analytics.is_enabled(): - return try: - wallet = await session.get(WalletModel, key.wallet_id) - root_id = key.wallet_id - if wallet is not None and wallet.parent_wallet_id: - root_id = wallet.parent_wallet_id - scope = select(col(WalletModel.wallet_id)).where( - or_( + root_id = await self._root_wallet_id(session, key.wallet_id) + claimed = await session.execute( + update(WalletModel) + .where( col(WalletModel.wallet_id) == root_id, - col(WalletModel.parent_wallet_id) == root_id, + col(WalletModel.product_activated_at).is_(None), ) + .values(product_activated_at=to_naive_utc(utc_now())) + .execution_options(synchronize_session=False) ) - prior = await session.execute( - select(func.count()) - .select_from(APIKeyModel) - .where( - col(APIKeyModel.key_id) != key.key_id, - col(APIKeyModel.first_used_at).is_not(None), - col(APIKeyModel.wallet_id).in_(scope), + await session.commit() + if (cast(Any, claimed).rowcount or 0) == 1: + product_analytics.track_activation( + root_id, key_id=key.key_id, wallet_id=key.wallet_id ) - ) - if (prior.scalar_one() or 0) > 0: - return - product_analytics.track_activation( - root_id, key_id=key.key_id, wallet_id=key.wallet_id - ) - except Exception as exc: # noqa: BLE001 - tracking must never raise - logger.debug( - "product_analytics_activation_failed error=%s", type(exc).__name__ - ) + except Exception as exc: # noqa: BLE001 - must never fail auth + await session.rollback() + logger.debug("product_activation_claim_failed error=%s", type(exc).__name__) + + @staticmethod + async def _root_wallet_id(session: AsyncSession, wallet_id: str) -> str: + """Walk parent_wallet_id up to the root (bounded, cycle-safe).""" + current = wallet_id + seen = {current} + for _ in range(_MAX_WALLET_DEPTH): + parent = ( + await session.execute( + select(col(WalletModel.parent_wallet_id)).where( + col(WalletModel.wallet_id) == current + ) + ) + ).scalar_one_or_none() + if not parent or parent in seen: + break + seen.add(parent) + current = parent + return current async def rotate_key( self, diff --git a/gateway/docs/product-analytics.md b/gateway/docs/product-analytics.md index 78eb9e8c..0be202f6 100644 --- a/gateway/docs/product-analytics.md +++ b/gateway/docs/product-analytics.md @@ -8,8 +8,9 @@ The gateway can send three product events to PostHog. It is off unless | `POSTHOG_API_KEY` | empty (off) | PostHog project API key. Blank or unset: every tracking call is a no-op. | | `POSTHOG_HOST` | `https://us.i.posthog.com` | PostHog ingestion host, e.g. `https://eu.i.posthog.com` or a self-hosted URL. | -Both are read from the process environment at send time (not from `Settings`, -so they never appear in a settings dump). Never commit a real key. +Each is read from the process environment first, then from `Settings`, so +`.env` entries work. The key is a `SecretStr` and is masked in settings dumps. +Never commit a real key. ## Events @@ -24,16 +25,22 @@ The distinct id is the account's root (sponsor) wallet id. Demo and dev self-provision signups are anonymous: no email is sent (dev wallets only carry a synthetic `@dev.local` address). -`activation` fires once per account. The key's first use is claimed with a -guarded `first_used_at IS NULL` update, so concurrent first uses produce one -event, and it is skipped when another key on the root wallet or its direct -child wallets was already used. Derived (JWT) authority recorded through +`activation` fires once per account. On a key's first use, the gateway walks +`parent_wallet_id` to the root wallet and claims +`wallets.product_activated_at` with a conditional `... IS NULL` UPDATE, so +concurrent first uses of one key or of several keys produce exactly one event. +The claim is recorded even when tracking is off, so turning tracking on later +doesn't replay old activations. Migration 054 backfills accounts that were +already active ([schema 054](schema-054-rollout.md)). `first_used_at` is still +set in the same single usage update as before. Derived (JWT) authority recorded through `consume_derived_key_use` is not an activation trigger. ## Failure behaviour Sends go to `/capture/` over httpx as background tasks with a -2 s timeout. Every error is caught and logged at debug level with the +2 s timeout. They reuse one shared httpx client and are capped at 256 in-flight +tasks: beyond that, events are dropped. App shutdown drains them and closes the +client. Every error is caught and logged at debug level with the exception type only (never the key), so tracking cannot fail or slow a request. Code lives in `app/core/product_analytics.py`; tests in `tests/test_product_analytics.py` mock the HTTP send. diff --git a/gateway/docs/schema-054-rollout.md b/gateway/docs/schema-054-rollout.md new file mode 100644 index 00000000..63144fbc --- /dev/null +++ b/gateway/docs/schema-054-rollout.md @@ -0,0 +1,25 @@ +# Schema 054: wallet product activation claim + +Migration `054_wallet_product_activation` revises `053_repair_api_lineage`. It +adds nullable `wallets.product_activated_at`, the account-level claim behind the +once-per-account PostHog `activation` event (see +[product analytics](product-analytics.md)). The gateway sets the column with a +conditional UPDATE (`WHERE product_activated_at IS NULL`) on a key's first use, +whether or not tracking is enabled. + +The upgrade backfills root wallets whose keys, or whose descendants' keys up to +three levels, were already used. Each gets its earliest `first_used_at`, so +existing accounts emit no late activation. Only NULL rows are updated. + +## Compatibility release + +Production boot requires the database revision to match the packaged head, so +a 053-packaged image cannot restart after 054 is applied. Qualify a +054-capable image against a disposable PostgreSQL upgrade from 053 first. +These release checks have not been run against staging or production. + +## Rollback boundary + +Downgrade drops the column. It holds only an analytics marker, so no signed or +financial data is lost. After a downgrade and re-upgrade, the backfill +re-derives the marker from `api_keys.first_used_at`. diff --git a/gateway/migrations/versions/054_wallet_product_activation.py b/gateway/migrations/versions/054_wallet_product_activation.py new file mode 100644 index 00000000..931ce3f6 --- /dev/null +++ b/gateway/migrations/versions/054_wallet_product_activation.py @@ -0,0 +1,53 @@ +"""Account-level product activation claim on wallets. + +Revision ID: 054_wallet_product_activation +Revises: 053_repair_api_lineage + +Adds nullable ``wallets.product_activated_at``. The gateway sets it once per +root wallet with a conditional UPDATE (``... IS NULL``) on a key's first use, +which makes the PostHog ``activation`` event fire once per account even when +first uses race. + +Backfill: a root wallet whose own keys, or whose descendants' keys (up to +three levels), were already used gets the earliest such ``first_used_at``, so +accounts active before 054 never emit a late activation. The column is added +only when missing, and the backfill only touches NULL rows. +""" + +import sqlalchemy as sa +from alembic import op + +revision: str = "054_wallet_product_activation" +down_revision: str = "053_repair_api_lineage" +branch_labels = None +depends_on = None + +BACKFILL_SQL = """ +UPDATE wallets SET product_activated_at = ( + SELECT MIN(k.first_used_at) + FROM api_keys k + JOIN wallets w1 ON w1.wallet_id = k.wallet_id + LEFT JOIN wallets w2 ON w2.wallet_id = w1.parent_wallet_id + LEFT JOIN wallets w3 ON w3.wallet_id = w2.parent_wallet_id + WHERE k.first_used_at IS NOT NULL + AND wallets.wallet_id IN (w1.wallet_id, w1.parent_wallet_id, + w2.parent_wallet_id, w3.parent_wallet_id) +) +WHERE parent_wallet_id IS NULL AND product_activated_at IS NULL +""" + + +def upgrade() -> None: + existing = {c["name"] for c in sa.inspect(op.get_bind()).get_columns("wallets")} + if "product_activated_at" not in existing: + with op.batch_alter_table("wallets") as batch_op: + batch_op.add_column( + sa.Column("product_activated_at", sa.DateTime(), nullable=True) + ) + op.execute(sa.text(BACKFILL_SQL)) + + +def downgrade() -> None: + # Analytics-only marker; dropping it loses no signed or financial data. + with op.batch_alter_table("wallets") as batch_op: + batch_op.drop_column("product_activated_at") diff --git a/gateway/tests/test_product_analytics.py b/gateway/tests/test_product_analytics.py index acabd513..81741eae 100644 --- a/gateway/tests/test_product_analytics.py +++ b/gateway/tests/test_product_analytics.py @@ -209,3 +209,133 @@ async def test_first_use_recorded_when_tracking_disabled( key = await service.validate_key(body["api_key"], consume=False) assert key.first_used_at is not None assert sent == [] + + +async def test_concurrent_first_uses_of_two_keys_fire_one_activation( + client, enabled, sent, self_provision +): + body = (await client.post("/v1/dev-keys/self-provision", json={})).json() + service = get_api_key_service() + second = await service.create_key(wallet_id=body["wallet_id"], key_name="2nd") + results = await asyncio.gather( + service.validate_key(body["api_key"]), + service.validate_key(second["api_key"]), + service.validate_key(body["api_key"]), + service.validate_key(second["api_key"]), + ) + assert all(results) + await product_analytics.drain_pending() + activations = [p for _, p in sent if p["event"] == "activation"] + assert len(activations) == 1 + assert activations[0]["distinct_id"] == body["sponsor_wallet_id"] + + +async def test_activation_claim_recorded_on_root_even_when_tracking_off( + client, monkeypatch, sent, self_provision +): + from app.db.database import get_session_factory + from app.db.models import WalletModel + + monkeypatch.delenv("POSTHOG_API_KEY", raising=False) + body = (await client.post("/v1/dev-keys/self-provision", json={})).json() + assert await get_api_key_service().validate_key(body["api_key"]) + async with get_session_factory()() as session: + root = await session.get(WalletModel, body["sponsor_wallet_id"]) + assert root.product_activated_at is not None + assert sent == [] + + +async def test_activation_claim_failure_does_not_fail_auth( + client, enabled, sent, self_provision, monkeypatch +): + from app.services.api_key_service import APIKeyService + + async def boom(session, wallet_id): + raise RuntimeError("db hiccup") + + monkeypatch.setattr(APIKeyService, "_root_wallet_id", staticmethod(boom)) + body = (await client.post("/v1/dev-keys/self-provision", json={})).json() + service = get_api_key_service() + assert await service.validate_key(body["api_key"]) is not None + reread = await service.validate_key(body["api_key"], consume=False) + assert reread.first_used_at is not None + await product_analytics.drain_pending() + assert "activation" not in _events(sent) + + +async def test_root_wallet_walk_handles_nesting_and_cycles(clean_database): + from app.db.database import get_session_factory + from app.db.models import WalletModel + from app.services.api_key_service import APIKeyService + + async with get_session_factory()() as session: + session.add(WalletModel(wallet_id="w-root", wallet_type="sponsor")) + await session.commit() + session.add( + WalletModel( + wallet_id="w-mid", wallet_type="agent", parent_wallet_id="w-root" + ) + ) + await session.commit() + session.add( + WalletModel( + wallet_id="w-leaf", wallet_type="agent", parent_wallet_id="w-mid" + ) + ) + await session.commit() + assert await APIKeyService._root_wallet_id(session, "w-leaf") == "w-root" + assert await APIKeyService._root_wallet_id(session, "w-root") == "w-root" + assert await APIKeyService._root_wallet_id(session, "missing") == "missing" + + +def test_posthog_settings_load_from_dotenv_fields(monkeypatch): + from app.core.config import Settings + + monkeypatch.delenv("POSTHOG_API_KEY", raising=False) + monkeypatch.delenv("POSTHOG_HOST", raising=False) + s = Settings( + _env_file=None, + POSTHOG_API_KEY=FAKE_KEY, + POSTHOG_HOST="https://eu.i.posthog.com", + ) + assert s.POSTHOG_API_KEY.get_secret_value() == FAKE_KEY + assert FAKE_KEY not in repr(s) + monkeypatch.setattr(get_settings(), "POSTHOG_API_KEY", s.POSTHOG_API_KEY) + monkeypatch.setattr(get_settings(), "POSTHOG_HOST", s.POSTHOG_HOST) + assert product_analytics._config() == (FAKE_KEY, "https://eu.i.posthog.com") + + +def test_settings_accepts_posthog_entries_in_env_file(tmp_path, monkeypatch): + from app.core.config import Settings + + env = tmp_path / ".env" + env.write_text( + f"POSTHOG_API_KEY={FAKE_KEY}\nPOSTHOG_HOST=https://eu.i.posthog.com\n" + ) + monkeypatch.delenv("POSTHOG_API_KEY", raising=False) + monkeypatch.delenv("POSTHOG_HOST", raising=False) + s = Settings(_env_file=str(env)) + assert s.POSTHOG_HOST == "https://eu.i.posthog.com" + + +async def test_pending_sends_are_capped(enabled, monkeypatch): + gate = asyncio.Event() + + async def slow(host, payload): + await gate.wait() + + monkeypatch.setattr(product_analytics, "_post", slow) + monkeypatch.setattr(product_analytics, "MAX_PENDING", 3) + for _ in range(10): + product_analytics.capture("signup", "w1") + assert len(product_analytics._pending) == 3 + gate.set() + await product_analytics.drain_pending() + + +async def test_shutdown_drains_and_closes_shared_client(enabled, monkeypatch): + client = product_analytics._get_client() + assert product_analytics._get_client() is client # reused + await product_analytics.shutdown() + assert client.is_closed + assert product_analytics._client is None From 95c4299f6c02f495ecf1f226e44601e894ac6f66 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Mon, 26 Oct 2026 07:10:43 -0700 Subject: [PATCH 3/4] fix(analytics): real two-key race test, unbounded-safe root resolution, recursive 054 backfill (#245) - Two-key activation test forces a lockstep interleave via a cyclic barrier on AsyncSession.commit; fails 3/3 on the 85ba83f logic. - _root_wallet_id returns None (claim skipped) on cycle, depth > cap, or an unreadable row (RLS child scope) instead of an intermediate wallet. - 054 backfill is one recursive CTE to the runtime depth (64), aggregated per wallet and walked by primary key; constant parity is tested. - Tests count cycle lookups and patch the depth cap; rollout note covers RLS and why the claim write stays on with tracking off. Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- gateway/app/services/api_key_service.py | 34 +++- gateway/docs/schema-054-rollout.md | 32 +++- .../versions/054_wallet_product_activation.py | 47 +++-- gateway/tests/test_product_analytics.py | 171 +++++++++++++++++- 4 files changed, 253 insertions(+), 31 deletions(-) diff --git a/gateway/app/services/api_key_service.py b/gateway/app/services/api_key_service.py index 7120c123..50f1349f 100644 --- a/gateway/app/services/api_key_service.py +++ b/gateway/app/services/api_key_service.py @@ -42,7 +42,9 @@ # Wallets nest via parent_wallet_id (sponsor -> agent -> child agent); the # root walk is capped so a corrupt cycle cannot loop forever. -_MAX_WALLET_DEPTH = 16 +# Bound on parent_wallet_id hops when resolving an account root. Shared with +# migration 054's backfill (054_wallet_product_activation.MAX_WALLET_DEPTH). +_MAX_WALLET_DEPTH = 64 API_KEY_LENGTH = 32 API_KEY_PREFIX_LENGTH = 8 @@ -568,6 +570,9 @@ async def _claim_account_activation( """ try: root_id = await self._root_wallet_id(session, key.wallet_id) + if root_id is None: + logger.debug("product_activation_root_unresolved") + return claimed = await session.execute( update(WalletModel) .where( @@ -587,23 +592,34 @@ async def _claim_account_activation( logger.debug("product_activation_claim_failed error=%s", type(exc).__name__) @staticmethod - async def _root_wallet_id(session: AsyncSession, wallet_id: str) -> str: - """Walk parent_wallet_id up to the root (bounded, cycle-safe).""" + async def _root_wallet_id(session: AsyncSession, wallet_id: str) -> str | None: + """Walk parent_wallet_id to the root, or None if it cannot be proven. + + Returns None (and the caller skips the claim) when the chain is deeper + than ``_MAX_WALLET_DEPTH``, contains a cycle, or reaches a row this + session cannot read (missing, or hidden by a child-scoped RLS policy). + Never returns an intermediate wallet as the account. + """ current = wallet_id seen = {current} - for _ in range(_MAX_WALLET_DEPTH): - parent = ( + for _ in range(_MAX_WALLET_DEPTH + 1): + row = ( await session.execute( select(col(WalletModel.parent_wallet_id)).where( col(WalletModel.wallet_id) == current ) ) - ).scalar_one_or_none() - if not parent or parent in seen: - break + ).first() + if row is None: + return None + parent = row[0] + if not parent: + return current + if parent in seen: + return None seen.add(parent) current = parent - return current + return None async def rotate_key( self, diff --git a/gateway/docs/schema-054-rollout.md b/gateway/docs/schema-054-rollout.md index 63144fbc..e9b423dc 100644 --- a/gateway/docs/schema-054-rollout.md +++ b/gateway/docs/schema-054-rollout.md @@ -7,9 +7,35 @@ once-per-account PostHog `activation` event (see conditional UPDATE (`WHERE product_activated_at IS NULL`) on a key's first use, whether or not tracking is enabled. -The upgrade backfills root wallets whose keys, or whose descendants' keys up to -three levels, were already used. Each gets its earliest `first_used_at`, so -existing accounts emit no late activation. Only NULL rows are updated. +The upgrade backfills root wallets whose keys, or any descendant's keys, were +already used. It walks ancestry with one recursive CTE to the same 64-hop +depth the runtime resolver uses (`_MAX_WALLET_DEPTH`; a test asserts they +match). `api_keys` is aggregated once per wallet and parents are followed by +the `wallets` primary key, so cost scales with used wallets times depth rather +than roots times keys. Each root gets its earliest `first_used_at`, so existing +accounts emit no late activation. Only NULL rows are updated, and cycles are +bounded by the depth limit and never reach a root. + +At runtime the resolver returns no account (and skips the claim, so no event) +when the chain is deeper than 64, has a cycle, or reaches a row the session +cannot read. It never claims an intermediate wallet. + +## Tenant RLS + +With `TENANT_RLS_MODE` on, a request session scoped to a child wallet may not +be able to read or update its root wallet row. The resolver then cannot prove +the root and skips the claim: activation is not recorded for that first use +and no event fires. It fails safe (no wrong account, no auth impact). That is +harmless while RLS is off by default; enabling RLS needs a privileged claim +path before activation counts can be trusted. + +## Claim write with tracking off + +The claim UPDATE runs even when PostHog is not configured. It is one +conditional UPDATE per key lifetime (first use only). Skipping it would leave +accounts that activated while tracking was off unmarked, and they would emit a +late `activation` once tracking is enabled, which is the inconsistency this +column exists to prevent. The write is kept on purpose. ## Compatibility release diff --git a/gateway/migrations/versions/054_wallet_product_activation.py b/gateway/migrations/versions/054_wallet_product_activation.py index 931ce3f6..348e82cf 100644 --- a/gateway/migrations/versions/054_wallet_product_activation.py +++ b/gateway/migrations/versions/054_wallet_product_activation.py @@ -8,9 +8,11 @@ which makes the PostHog ``activation`` event fire once per account even when first uses race. -Backfill: a root wallet whose own keys, or whose descendants' keys (up to -three levels), were already used gets the earliest such ``first_used_at``, so -accounts active before 054 never emit a late activation. The column is added +Backfill: a root wallet whose own keys, or any descendant's keys (to the same +``MAX_WALLET_DEPTH`` the runtime resolver uses), were already used gets the +earliest such ``first_used_at``, so accounts active before 054 never emit a +late activation. One recursive CTE: api_keys is aggregated once per wallet, +then ancestry is walked via the wallets primary key (no per-root scan). The column is added only when missing, and the backfill only touches NULL rows. """ @@ -22,18 +24,37 @@ branch_labels = None depends_on = None -BACKFILL_SQL = """ +# Must match app.services.api_key_service._MAX_WALLET_DEPTH (asserted by a test). +MAX_WALLET_DEPTH = 64 + +BACKFILL_SQL = f""" +WITH RECURSIVE used AS ( + SELECT wallet_id, MIN(first_used_at) AS first_used_at + FROM api_keys + WHERE first_used_at IS NOT NULL + GROUP BY wallet_id +), +anc (wallet_id, ancestor_id, depth) AS ( + SELECT wallet_id, wallet_id, 0 FROM used + UNION ALL + SELECT a.wallet_id, w.parent_wallet_id, a.depth + 1 + FROM anc a + JOIN wallets w ON w.wallet_id = a.ancestor_id + WHERE w.parent_wallet_id IS NOT NULL AND a.depth < {MAX_WALLET_DEPTH} +), +root_first AS ( + SELECT a.ancestor_id AS root_id, MIN(u.first_used_at) AS first_used_at + FROM anc a + JOIN used u ON u.wallet_id = a.wallet_id + JOIN wallets r ON r.wallet_id = a.ancestor_id AND r.parent_wallet_id IS NULL + GROUP BY a.ancestor_id +) UPDATE wallets SET product_activated_at = ( - SELECT MIN(k.first_used_at) - FROM api_keys k - JOIN wallets w1 ON w1.wallet_id = k.wallet_id - LEFT JOIN wallets w2 ON w2.wallet_id = w1.parent_wallet_id - LEFT JOIN wallets w3 ON w3.wallet_id = w2.parent_wallet_id - WHERE k.first_used_at IS NOT NULL - AND wallets.wallet_id IN (w1.wallet_id, w1.parent_wallet_id, - w2.parent_wallet_id, w3.parent_wallet_id) + SELECT rf.first_used_at FROM root_first rf WHERE rf.root_id = wallets.wallet_id ) -WHERE parent_wallet_id IS NULL AND product_activated_at IS NULL +WHERE parent_wallet_id IS NULL + AND product_activated_at IS NULL + AND wallet_id IN (SELECT root_id FROM root_first) """ diff --git a/gateway/tests/test_product_analytics.py b/gateway/tests/test_product_analytics.py index 81741eae..c61c21d9 100644 --- a/gateway/tests/test_product_analytics.py +++ b/gateway/tests/test_product_analytics.py @@ -212,18 +212,60 @@ async def test_first_use_recorded_when_tracking_disabled( async def test_concurrent_first_uses_of_two_keys_fire_one_activation( - client, enabled, sent, self_provision + client, enabled, sent, self_provision, monkeypatch ): + """Two keys of one account are first used truly concurrently. + + SQLite serializes plain ``asyncio.gather`` calls, so a barrier is patched + into ``AsyncSession.commit``: each racing call commits its key's usage + update and then waits until the other has done the same. Both first uses + are therefore durable before either call decides on activation, which is + exactly the interleave that broke the old post-commit "count other used + keys" logic (each saw the other and neither fired). The atomic claim on + the root wallet must still fire exactly once. + """ + import contextvars + + from sqlalchemy.ext.asyncio import AsyncSession + body = (await client.post("/v1/dev-keys/self-provision", json={})).json() service = get_api_key_service() second = await service.create_key(wallet_id=body["wallet_id"], key_name="2nd") + + # Cyclic barrier: every commit of a racing call waits for the other's + # matching commit, so the two calls advance in lockstep (usage commit, + # then claim commit) and both first uses are durable before either + # decides. A call that finishes early just releases the barrier (timeout). + barrier = asyncio.Barrier(2) + racing: contextvars.ContextVar[bool] = contextvars.ContextVar( + "racing", default=False + ) + lockstep = {"on": True, "rounds": 0} + original_commit = AsyncSession.commit + + async def commit_then_wait(self): + await original_commit(self) + if racing.get() and lockstep["on"]: + try: + await asyncio.wait_for(barrier.wait(), timeout=1) + lockstep["rounds"] += 1 + except TimeoutError: + lockstep["on"] = False + await barrier.reset() + + monkeypatch.setattr(AsyncSession, "commit", commit_then_wait) + + async def race(api_key: str): + racing.set(True) + return await service.validate_key(api_key) + results = await asyncio.gather( - service.validate_key(body["api_key"]), - service.validate_key(second["api_key"]), - service.validate_key(body["api_key"]), - service.validate_key(second["api_key"]), + asyncio.create_task(race(body["api_key"])), + asyncio.create_task(race(second["api_key"])), ) + monkeypatch.setattr(AsyncSession, "commit", original_commit) assert all(results) + assert lockstep["rounds"] >= 2 # both usage commits met before deciding await product_analytics.drain_pending() activations = [p for _, p in sent if p["event"] == "activation"] assert len(activations) == 1 @@ -285,7 +327,124 @@ async def test_root_wallet_walk_handles_nesting_and_cycles(clean_database): await session.commit() assert await APIKeyService._root_wallet_id(session, "w-leaf") == "w-root" assert await APIKeyService._root_wallet_id(session, "w-root") == "w-root" - assert await APIKeyService._root_wallet_id(session, "missing") == "missing" + # An unreadable row (missing, or hidden by RLS) is never treated as root. + assert await APIKeyService._root_wallet_id(session, "missing") is None + + +async def _chain(session, ids, *, parent_of_first=None): + from app.db.models import WalletModel + + parent = parent_of_first + for wid in ids: + session.add( + WalletModel(wallet_id=wid, wallet_type="agent", parent_wallet_id=parent) + ) + await session.commit() + parent = wid + + +async def test_root_wallet_walk_cycle_is_unresolved_and_bounded( + clean_database, monkeypatch +): + from sqlalchemy.ext.asyncio import AsyncSession + + from app.db.database import get_session_factory + from app.db.models import WalletModel + from app.services.api_key_service import APIKeyService + + async with get_session_factory()() as session: + await _chain(session, ["cyc-a", "cyc-b"]) + a = await session.get(WalletModel, "cyc-a") + a.parent_wallet_id = "cyc-b" + await session.commit() + + calls = {"n": 0} + original = AsyncSession.execute + + async def counting(self, *args, **kwargs): + calls["n"] += 1 + return await original(self, *args, **kwargs) + + monkeypatch.setattr(AsyncSession, "execute", counting) + assert await APIKeyService._root_wallet_id(session, "cyc-a") is None + monkeypatch.setattr(AsyncSession, "execute", original) + # The seen-set stops the cycle after visiting both wallets, well + # before the depth cap would. + assert calls["n"] == 2 + + +async def test_root_wallet_walk_depth_cap_is_unresolved(clean_database, monkeypatch): + from app.db.database import get_session_factory + from app.services import api_key_service as svc + + async with get_session_factory()() as session: + ids = [f"deep-{i}" for i in range(6)] + await _chain(session, ids) + leaf = ids[-1] + assert await svc.APIKeyService._root_wallet_id(session, leaf) == "deep-0" + monkeypatch.setattr(svc, "_MAX_WALLET_DEPTH", 3) + # Deeper than the cap: refuse rather than return an intermediate wallet. + assert await svc.APIKeyService._root_wallet_id(session, leaf) is None + assert await svc.APIKeyService._root_wallet_id(session, "deep-3") == "deep-0" + + +def test_backfill_depth_matches_runtime(): + import importlib.util + from pathlib import Path + + from app.services import api_key_service as svc + + path = ( + Path(__file__).resolve().parents[1] + / "migrations/versions/054_wallet_product_activation.py" + ) + spec = importlib.util.spec_from_file_location("m054", path) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + assert mod.MAX_WALLET_DEPTH == svc._MAX_WALLET_DEPTH + + +async def test_backfill_marks_root_of_deep_used_descendant(clean_database): + import importlib.util + from datetime import datetime + from pathlib import Path + + from sqlalchemy import text + + from app.db.database import get_session_factory + from app.db.models import APIKeyModel, WalletModel + + path = ( + Path(__file__).resolve().parents[1] + / "migrations/versions/054_wallet_product_activation.py" + ) + spec = importlib.util.spec_from_file_location("m054b", path) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + + used_at = datetime(2026, 1, 2, 3, 4, 5) + async with get_session_factory()() as session: + ids = [f"bf-{i}" for i in range(6)] # depth 5, beyond the old 3 levels + await _chain(session, ids) + session.add(WalletModel(wallet_id="bf-unused-root", wallet_type="sponsor")) + session.add( + APIKeyModel( + key_id="bf-key", + wallet_id=ids[-1], + key_hash="x" * 64, + key_prefix="b2a_bfbf", + first_used_at=used_at, + ) + ) + await session.commit() + await session.execute(text(mod.BACKFILL_SQL)) + await session.commit() + session.expire_all() + root = await session.get(WalletModel, "bf-0") + assert root.product_activated_at == used_at + assert (await session.get(WalletModel, "bf-3")).product_activated_at is None + unused = await session.get(WalletModel, "bf-unused-root") + assert unused.product_activated_at is None def test_posthog_settings_load_from_dotenv_fields(monkeypatch): From 22fe3ba805a5c6ac3423bf28ec1d38910d4572a8 Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Mon, 26 Oct 2026 21:54:46 -0700 Subject: [PATCH 4/4] fix(analytics): require HTTPS PostHog host; document migration 055 modes Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- CHANGELOG.md | 4 +++- gateway/app/core/product_analytics.py | 4 ++++ gateway/tests/test_product_analytics.py | 11 +++++++++++ 3 files changed, 18 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e3674e77..5c1ef003 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,9 @@ This repository starts with a fresh history. Releases are listed here once tagge docs/api-key-hashing.md. - Optional PostHog product tracking (`POSTHOG_API_KEY`, `POSTHOG_HOST`; off when unset): identify with email and `signup` on account creation, `activation` once - per account on first key use. Fail-safe and non-blocking. + per account on first key use. Fail-safe and non-blocking. Adds migration 055: run + `alembic upgrade head` before starting the image, or set + `RUN_MIGRATIONS_ON_START=true` so the entrypoint migrates on boot. - Upstream MCP (Preview): optional multi-tool discovery and exact-name allowlist with stable public IDs, per-tool permit scope, and unchanged per-call credits. Health reports `upstream_tool_mode: multi_tool_preview` while it is on. diff --git a/gateway/app/core/product_analytics.py b/gateway/app/core/product_analytics.py index 64a1884b..983ed6b7 100644 --- a/gateway/app/core/product_analytics.py +++ b/gateway/app/core/product_analytics.py @@ -54,6 +54,10 @@ def _config() -> tuple[str, str] | None: or settings.POSTHOG_HOST.strip() or DEFAULT_HOST ) + if not host.lower().startswith("https://"): + # The identify payload carries the sponsor email; never send it in + # cleartext. A non-HTTPS host leaves tracking off. + return None return api_key, host.rstrip("/") diff --git a/gateway/tests/test_product_analytics.py b/gateway/tests/test_product_analytics.py index 936b87b8..139d8ee0 100644 --- a/gateway/tests/test_product_analytics.py +++ b/gateway/tests/test_product_analytics.py @@ -498,3 +498,14 @@ async def test_shutdown_drains_and_closes_shared_client(enabled, monkeypatch): await product_analytics.shutdown() assert client.is_closed assert product_analytics._client is None + + +@pytest.mark.anyio +@pytest.mark.parametrize("host", ["http://posthog.internal", "ftp://x", "posthog.com"]) +async def test_non_https_host_disables_tracking(monkeypatch, sent, host): + monkeypatch.setenv("POSTHOG_API_KEY", FAKE_KEY) + monkeypatch.setenv("POSTHOG_HOST", host) + assert product_analytics.is_enabled() is False + product_analytics.capture("signup", "w1") + await product_analytics.drain_pending() + assert sent == []