fix: attribute PostHog events to automation actors

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.

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")

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),

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()

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(

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,
     }
 

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)
 

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

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."""