diff --git a/AGENTS.md b/AGENTS.md index 319911b..27fdd86 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -105,6 +105,13 @@ OPENHANDS_API_KEY=sk-oh-... uv run pytest tests/integration/ -v OPENHANDS_API_KEY=sk-oh-... uv run python scripts/test_automation.py --api-url https://staging.all-hands.dev ``` +## Product Telemetry Identity + +- Cloud product events use the server-authoritative Cloud user ID as the PostHog `distinct_id`; never trust a client telemetry header as Cloud identity. +- Local Agent Canvas requests carry their consented PostHog identity in `X-OpenHands-Telemetry-Distinct-Id`. Store that value on newly created automations and runs so asynchronous lifecycle events retain the same identity. +- Local events with no Canvas attribution fall back to the DB-backed automation backend ID. Keep `automation_backend_id` as an event property for installation analysis rather than substituting it for a known person identity. +- Local consent is stored per frontend distinct ID. Attributed events require consent for their resolved identity; only unattributed backend-level events use the aggregate installation consent. + ## PR-Specific Documents When working on a PR that requires design documents, live-test logs, development-only scripts, or other temporary artifacts that should **not** be merged to `main`, store them in a `.pr/` directory at the repository root. diff --git a/migrations/versions/012_add_telemetry_attribution.py b/migrations/versions/012_add_telemetry_attribution.py new file mode 100644 index 0000000..d1de365 --- /dev/null +++ b/migrations/versions/012_add_telemetry_attribution.py @@ -0,0 +1,33 @@ +"""Add telemetry attribution to automations and runs. + +Revision ID: 012 +Revises: 011 +Create Date: 2026-07-26 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + + +revision: str = "012" +down_revision: str = "011" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "automations", + sa.Column("telemetry_distinct_id", sa.String(length=256), nullable=True), + ) + op.add_column( + "automation_runs", + sa.Column("telemetry_distinct_id", sa.String(length=256), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("automation_runs", "telemetry_distinct_id") + op.drop_column("automations", "telemetry_distinct_id") diff --git a/openhands/automation/models.py b/openhands/automation/models.py index d98408e..888ef56 100644 --- a/openhands/automation/models.py +++ b/openhands/automation/models.py @@ -53,6 +53,9 @@ class Automation(Base): user_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False, index=True) org_id: Mapped[uuid.UUID] = mapped_column(Uuid, nullable=False, index=True) name: Mapped[str] = mapped_column(String(500), nullable=False) + telemetry_distinct_id: Mapped[str | None] = mapped_column( + String(256), nullable=True + ) # Optional prompt (set when created via preset endpoints) prompt: Mapped[str | None] = mapped_column(Text, nullable=True) @@ -134,6 +137,9 @@ class AutomationRun(Base): nullable=False, index=True, ) + telemetry_distinct_id: Mapped[str | None] = mapped_column( + String(256), nullable=True + ) status: Mapped[AutomationRunStatus] = mapped_column( Enum(AutomationRunStatus, native_enum=False, length=20), diff --git a/openhands/automation/preset_router.py b/openhands/automation/preset_router.py index 492b6d2..4dc5f56 100644 --- a/openhands/automation/preset_router.py +++ b/openhands/automation/preset_router.py @@ -30,7 +30,10 @@ from openhands.automation.models import Automation, TarballUpload, UploadStatus from openhands.automation.schemas import AutomationResponse, Trigger from openhands.automation.storage import FileStore, get_file_store -from openhands.automation.telemetry import capture_automation_event +from openhands.automation.telemetry import ( + capture_automation_event, + get_request_telemetry_context, +) from openhands.automation.utils import utcnow from openhands.automation.utils.model_profiles import resolve_model_profile_for_user from openhands.automation.utils.tarball_validation import ( @@ -464,6 +467,9 @@ async def create_automation_from_prompt( entrypoint=_get_preset_entrypoint(), timeout=default_automation_timeout(body.timeout), keep_alive=body.keep_alive, + telemetry_distinct_id=get_request_telemetry_context( + request + ).frontend_distinct_id, ) session.add(automation) await session.flush() @@ -845,6 +851,9 @@ async def create_automation_from_plugin( entrypoint=_get_preset_entrypoint(), timeout=default_automation_timeout(body.timeout), keep_alive=body.keep_alive, + telemetry_distinct_id=get_request_telemetry_context( + request + ).frontend_distinct_id, ) session.add(automation) await session.flush() diff --git a/openhands/automation/router.py b/openhands/automation/router.py index 1b9d7f9..4fc7ca1 100644 --- a/openhands/automation/router.py +++ b/openhands/automation/router.py @@ -31,7 +31,10 @@ UpdateAutomationRequest, ) from openhands.automation.storage import FileStore, get_file_store -from openhands.automation.telemetry import capture_automation_event +from openhands.automation.telemetry import ( + capture_automation_event, + get_request_telemetry_context, +) from openhands.automation.utils import utcnow from openhands.automation.utils.api_key import ( APIKeyError, @@ -91,6 +94,9 @@ async def create_automation( entrypoint=body.entrypoint, timeout=default_automation_timeout(body.timeout), keep_alive=body.keep_alive, + telemetry_distinct_id=get_request_telemetry_context( + request + ).frontend_distinct_id, ) session.add(auto) await session.flush() @@ -301,7 +307,13 @@ async def dispatch_automation( picked up by the dispatcher and executed. """ auto = await _get_user_automation(session, automation_id, user.user_id, user.org_id) - run = await create_pending_run(session, auto) + run = await create_pending_run( + session, + auto, + telemetry_distinct_id=get_request_telemetry_context( + request + ).frontend_distinct_id, + ) await session.flush() await session.refresh(run) await capture_automation_event( diff --git a/openhands/automation/telemetry.py b/openhands/automation/telemetry.py index e282119..58ee417 100644 --- a/openhands/automation/telemetry.py +++ b/openhands/automation/telemetry.py @@ -151,13 +151,20 @@ async def get_stored_telemetry_consent( request: Request | None = None, session: AsyncSession | None = None, session_factory: async_sessionmaker[AsyncSession] | None = None, + frontend_distinct_id: str | None = None, ) -> bool: - """Return whether any known frontend identity granted telemetry consent.""" + """Return aggregate consent or consent for a specific frontend identity.""" + + def resolve(consents: dict[str, bool]) -> bool: + if frontend_distinct_id is None: + return has_granted_telemetry_consent(consents) + return consents.get( + _normalize_frontend_distinct_id(frontend_distinct_id), False + ) + try: if session is not None: - return has_granted_telemetry_consent( - await _load_telemetry_consent_map(session) - ) + return resolve(await _load_telemetry_consent_map(session)) if session_factory is None and request is not None: session_factory = getattr(request.app.state, "session_factory", None) @@ -169,9 +176,7 @@ async def get_stored_telemetry_consent( return False async with session_factory() as new_session: - return has_granted_telemetry_consent( - await _load_telemetry_consent_map(new_session) - ) + return resolve(await _load_telemetry_consent_map(new_session)) except Exception: logger.debug("Failed to load automation telemetry consent", exc_info=True) return False @@ -181,9 +186,21 @@ def get_request_telemetry_context(request: Request | None) -> TelemetryRequestCo if request is None: return TelemetryRequestContext() context = getattr(request.state, "telemetry_context", None) - if isinstance(context, TelemetryRequestContext): + if not isinstance(context, TelemetryRequestContext): + context = build_telemetry_request_context(request.scope) + return _trusted_telemetry_context(context) + + +def _trusted_telemetry_context( + context: TelemetryRequestContext, +) -> TelemetryRequestContext: + """Discard browser identity where Cloud must derive identity from auth.""" + if get_config().service.is_local_mode: return context - return build_telemetry_request_context(request.scope) + return TelemetryRequestContext( + client_source=context.client_source, + client_version=context.client_version, + ) def get_request_authenticated_user(request: Request) -> AuthenticatedUser | None: @@ -254,8 +271,45 @@ def _trigger_type(automation: Automation | None) -> str | None: return str(trigger) if trigger is not None else None -def _resolve_distinct_id(*, backend_distinct_id: str) -> str: - return backend_distinct_id +def _resolve_local_frontend_distinct_id( + *, + request_context: TelemetryRequestContext, + automation: Automation | None, + run: AutomationRun | None, +) -> str | None: + if run is not None and run.telemetry_distinct_id: + return run.telemetry_distinct_id + if request_context.frontend_distinct_id: + return request_context.frontend_distinct_id + if automation is not None: + return automation.telemetry_distinct_id + return None + + +def _resolve_distinct_id( + *, + request_context: TelemetryRequestContext, + user: AuthenticatedUser | None, + automation: Automation | None, + run: AutomationRun | None, + backend_distinct_id: str, +) -> str: + settings = get_config().service + if not settings.is_local_mode: + if user is not None: + return str(user.user_id) + if automation is not None: + return str(automation.user_id) + return backend_distinct_id + + return ( + _resolve_local_frontend_distinct_id( + request_context=request_context, + automation=automation, + run=run, + ) + or backend_distinct_id + ) def _base_properties( @@ -344,11 +398,25 @@ async def capture_automation_event( if not settings.posthog_api_key: return - context = request_context or get_request_telemetry_context(request) + context = ( + _trusted_telemetry_context(request_context) + if request_context is not None + else get_request_telemetry_context(request) + ) + local_frontend_distinct_id = ( + _resolve_local_frontend_distinct_id( + request_context=context, + automation=automation, + run=run, + ) + if settings.is_local_mode + else None + ) if settings.is_local_mode and not await get_stored_telemetry_consent( request=request, session=session, session_factory=session_factory, + frontend_distinct_id=local_frontend_distinct_id, ): return @@ -373,7 +441,13 @@ async def capture_automation_event( payload = { "api_key": settings.posthog_api_key, "event": event, - "distinct_id": _resolve_distinct_id(backend_distinct_id=backend_distinct_id), + "distinct_id": _resolve_distinct_id( + request_context=context, + user=user, + automation=automation, + run=run, + backend_distinct_id=backend_distinct_id, + ), "properties": event_properties, } diff --git a/openhands/automation/utils/run.py b/openhands/automation/utils/run.py index 4a7aab6..b25a7bb 100644 --- a/openhands/automation/utils/run.py +++ b/openhands/automation/utils/run.py @@ -79,6 +79,8 @@ async def disable_automation( async def create_pending_run( session: AsyncSession, automation: Automation, + *, + telemetry_distinct_id: str | None = None, ) -> AutomationRun: """Create a PENDING automation run for dispatch. @@ -98,6 +100,9 @@ async def create_pending_run( id=uuid.uuid4(), automation_id=automation.id, status=AutomationRunStatus.PENDING, + telemetry_distinct_id=( + telemetry_distinct_id or automation.telemetry_distinct_id + ), ) session.add(run) diff --git a/openhands/automation/utils/webhook.py b/openhands/automation/utils/webhook.py index fa60b2d..6c1ba8f 100644 --- a/openhands/automation/utils/webhook.py +++ b/openhands/automation/utils/webhook.py @@ -300,6 +300,7 @@ async def create_automation_run( automation_id=automation.id, status=AutomationRunStatus.PENDING, event_payload=event_payload, + telemetry_distinct_id=automation.telemetry_distinct_id, ) session.add(run) return run diff --git a/tests/test_router.py b/tests/test_router.py index 9f8f09a..9910a8c 100644 --- a/tests/test_router.py +++ b/tests/test_router.py @@ -7,7 +7,12 @@ import pytest -from openhands.automation.models import Automation, TarballUpload, UploadStatus +from openhands.automation.models import ( + Automation, + AutomationRun, + TarballUpload, + UploadStatus, +) from openhands.automation.preset_router import _build_storage_path, _generate_tarball from openhands.automation.utils import utcnow from openhands.automation.utils.tarball_validation import ( @@ -168,7 +173,11 @@ async def test_create_automation_success(self, async_client, async_session): "entrypoint": "uv run script.py", } - response = await async_client.post("/api/automation/v1", json=payload) + response = await async_client.post( + "/api/automation/v1", + json=payload, + headers={"X-OpenHands-Telemetry-Distinct-Id": "ph-fe-creator"}, + ) assert response.status_code == 201 data = response.json() @@ -185,6 +194,9 @@ async def test_create_automation_success(self, async_client, async_session): assert data["enabled"] is True assert "id" in data assert data["user_id"] == str(TEST_USER_ID) + automation = await async_session.get(Automation, uuid.UUID(data["id"])) + assert automation is not None + assert automation.telemetry_distinct_id == "ph-fe-creator" async def test_create_automation_defaults_to_active_model_profile( self, async_client, mock_authenticated_user @@ -1213,7 +1225,8 @@ async def test_dispatch_automation_success(self, async_client, async_session): await async_session.commit() response = await async_client.post( - f"/api/automation/v1/{automation.id}/dispatch" + f"/api/automation/v1/{automation.id}/dispatch", + headers={"X-OpenHands-Telemetry-Distinct-Id": "ph-fe-dispatcher"}, ) assert response.status_code == 201 @@ -1225,6 +1238,9 @@ async def test_dispatch_automation_success(self, async_client, async_session): assert "created_at" in data assert data["started_at"] is None assert data["completed_at"] is None + run = await async_session.get(AutomationRun, uuid.UUID(data["id"])) + assert run is not None + assert run.telemetry_distinct_id == "ph-fe-dispatcher" async def test_dispatch_automation_not_found(self, async_client): """Dispatching a nonexistent automation returns 404.""" diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index dd41cba..320ba34 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -22,7 +22,9 @@ ) from openhands.automation.schemas import TelemetryConsentRequest from openhands.automation.telemetry_router import set_telemetry_consent +from openhands.automation.utils.run import create_pending_run from openhands.automation.utils.version import get_server_version_info +from openhands.automation.utils.webhook import create_automation_run class _Response: @@ -47,7 +49,7 @@ async def post(self, url: str, json: dict) -> _Response: return _Response() -def _automation() -> Automation: +def _automation(telemetry_distinct_id: str | None = None) -> Automation: return Automation( id=uuid.uuid4(), user_id=uuid.uuid4(), @@ -58,6 +60,7 @@ def _automation() -> Automation: entrypoint="python main.py", enabled=True, timeout=300, + telemetry_distinct_id=telemetry_distinct_id, ) @@ -91,7 +94,7 @@ def _reset_config(monkeypatch): @pytest.mark.asyncio -async def test_local_capture_uses_backend_id_and_frontend_property(monkeypatch): +async def test_local_capture_uses_canvas_distinct_id(monkeypatch): monkeypatch.setenv("AUTOMATION_POSTHOG_API_KEY", "ph_test") monkeypatch.setenv("AUTOMATION_POSTHOG_HOST", "https://posthog.example") monkeypatch.setenv("AUTOMATION_AGENT_SERVER_URL", "http://localhost:3000") @@ -128,7 +131,7 @@ async def stored_consent(**kwargs): url, payload = _MockAsyncClient.posts[0] assert url == "https://posthog.example/capture/" assert payload["event"] == "automation_run_completed" - assert payload["distinct_id"] == "automation-backend:test" + assert payload["distinct_id"] == "ph-fe-123" properties = payload["properties"] _assert_server_version_properties(properties) @@ -148,7 +151,7 @@ async def stored_consent(**kwargs): @pytest.mark.asyncio -async def test_cloud_capture_uses_backend_id_and_org_properties(monkeypatch): +async def test_cloud_capture_uses_user_id_and_org_properties(monkeypatch): monkeypatch.setenv("AUTOMATION_POSTHOG_API_KEY", "ph_test") clear_config_cache() monkeypatch.setattr(telemetry.httpx, "AsyncClient", _MockAsyncClient) @@ -168,7 +171,7 @@ async def backend_id(**kwargs): ) _, payload = _MockAsyncClient.posts[0] - assert payload["distinct_id"] == "automation-backend:cloud" + assert payload["distinct_id"] == str(automation.user_id) properties = payload["properties"] assert properties["deployment_mode"] == "cloud" assert properties["cloud_user_id"] == str(automation.user_id) @@ -177,7 +180,7 @@ async def backend_id(**kwargs): assert "org_id" not in properties assert properties["creation_path"] == "prompt_preset" - assert properties["frontend_distinct_id"] == "ph-fe-123" + assert "frontend_distinct_id" not in properties assert properties["automation_backend_id"] == "automation-backend:cloud" @@ -225,14 +228,15 @@ async def backend_id(**kwargs): monkeypatch.setattr(telemetry, "get_automation_backend_distinct_id", backend_id) + automation = _automation() await telemetry.capture_automation_event( "automation_created", - automation=_automation(), + automation=automation, ) assert len(_MockAsyncClient.posts) == 1 _, payload = _MockAsyncClient.posts[0] - assert payload["distinct_id"] == "automation-backend:cloud" + assert payload["distinct_id"] == str(automation.user_id) assert payload["properties"]["deployment_mode"] == "cloud" @@ -257,15 +261,21 @@ async def test_local_capture_uses_stored_consent_without_request_id(monkeypatch) ) await session.commit() + automation = _automation(telemetry_distinct_id="ph-fe-consented") + run = _run(automation) + run.telemetry_distinct_id = "ph-fe-consented" + await telemetry.capture_automation_event( "automation_run_dispatched", - automation=_automation(), + automation=automation, + run=run, session_factory=session_factory, ) assert len(_MockAsyncClient.posts) == 1 _, payload = _MockAsyncClient.posts[0] assert payload["event"] == "automation_run_dispatched" + assert payload["distinct_id"] == "ph-fe-consented" assert payload["properties"]["deployment_mode"] == "local" assert "frontend_distinct_id" not in payload["properties"] finally: @@ -292,6 +302,14 @@ async def test_stored_telemetry_consent_tracks_any_granted_frontend_id(): frontend_distinct_id="ph-fe-b", ) assert await telemetry.get_stored_telemetry_consent(session=session) + assert not await telemetry.get_stored_telemetry_consent( + session=session, + frontend_distinct_id="ph-fe-a", + ) + assert await telemetry.get_stored_telemetry_consent( + session=session, + frontend_distinct_id="ph-fe-b", + ) assert await telemetry.set_stored_telemetry_consent( session, @@ -311,6 +329,28 @@ async def test_stored_telemetry_consent_tracks_any_granted_frontend_id(): await engine.dispose() +@pytest.mark.asyncio +async def test_background_runs_inherit_automation_telemetry_identity(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + try: + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + session_factory = async_sessionmaker(engine, expire_on_commit=False) + async with session_factory() as session: + automation = _automation(telemetry_distinct_id="ph-fe-owner") + session.add(automation) + await session.flush() + + scheduled_run = await create_pending_run(session, automation) + webhook_run = await create_automation_run(automation, session) + + assert scheduled_run.telemetry_distinct_id == "ph-fe-owner" + assert webhook_run.telemetry_distinct_id == "ph-fe-owner" + finally: + await engine.dispose() + + @pytest.mark.asyncio async def test_telemetry_consent_route_stores_consent_and_emits_link_event( monkeypatch, @@ -359,7 +399,7 @@ async def test_telemetry_consent_route_stores_consent_and_emits_link_event( _, payload = _MockAsyncClient.posts[0] properties = payload["properties"] assert payload["event"] == "automation_telemetry_consent_granted" - assert payload["distinct_id"].startswith("automation-backend:") + assert payload["distinct_id"] == "ph-fe-link" assert properties["frontend_distinct_id"] == "ph-fe-link" assert properties["client_source"] == "agent_canvas" assert properties["client_version"] == "1.2.3" @@ -497,6 +537,10 @@ async def backend_id(**kwargs): monkeypatch.setattr(telemetry, "get_automation_backend_distinct_id", backend_id) request = _request("/api/automation/v1/123", endpoint_name="get_automation") + request.state.telemetry_context = TelemetryRequestContext( + frontend_distinct_id="untrusted-browser-id", + client_source="agent_canvas", + ) user = AuthenticatedUser( user_id=uuid.uuid4(), org_id=uuid.uuid4(), @@ -523,9 +567,12 @@ async def backend_id(**kwargs): _assert_server_version_properties(properties) assert properties["deployment_mode"] == "cloud" + assert payload["distinct_id"] == str(user.user_id) assert properties["cloud_user_id"] == str(user.user_id) assert properties["cloud_org_id"] == str(user.org_id) assert properties["$groups"] == {"org": str(user.org_id)} + assert "frontend_distinct_id" not in properties + assert properties["client_source"] == "agent_canvas" assert "org_id" not in properties assert properties["success"] is True assert properties["duration_ms"] == 12 @@ -571,7 +618,7 @@ async def stored_consent(**kwargs): _, payload = _MockAsyncClient.posts[0] properties = payload["properties"] - assert payload["distinct_id"] == "automation-backend:local-api" + assert payload["distinct_id"] == "ph-fe-local" assert properties["deployment_mode"] == "local" assert properties["automation_backend_id"] == "automation-backend:local-api" assert "cloud_user_id" not in properties @@ -583,6 +630,31 @@ async def stored_consent(**kwargs): assert properties["duration_ms"] == 12 +@pytest.mark.asyncio +async def test_local_capture_falls_back_to_backend_id_without_canvas_actor( + monkeypatch, +): + monkeypatch.setenv("AUTOMATION_POSTHOG_API_KEY", "ph_test") + monkeypatch.setenv("AUTOMATION_AGENT_SERVER_URL", "http://localhost:3000") + clear_config_cache() + monkeypatch.setattr(telemetry.httpx, "AsyncClient", _MockAsyncClient) + + async def backend_id(**kwargs): + return "automation-backend:local-api" + + monkeypatch.setattr(telemetry, "get_automation_backend_distinct_id", backend_id) + + async def stored_consent(**kwargs): + return True + + monkeypatch.setattr(telemetry, "get_stored_telemetry_consent", stored_consent) + + await telemetry.capture_automation_event("automation_event_received") + + _, payload = _MockAsyncClient.posts[0] + assert payload["distinct_id"] == "automation-backend:local-api" + + @pytest.mark.asyncio async def test_backend_distinct_id_is_db_backed_and_stable(): engine = create_async_engine("sqlite+aiosqlite:///:memory:")