feat: emit canonical conversation telemetry from agent server

openhands-agent-server/openhands/agent_server/README.md

@@ -194,7 +194,7 @@ repository.
 
 #### What is sent
 
-Events: `conversation_started`, `conversation_finished`, `conversation_failed`,
+Events: `conversation_created`, `conversation_finished`, `conversation_failed`,
 `conversation_error`, `request_failed`, `server_started`, `server_stopped` — all
 prefixed `agent_server.`.
 

openhands-agent-server/openhands/agent_server/conversation_service.py

@@ -2127,10 +2127,10 @@ async def _maybe_subscribe_telemetry(
 
         The subscriber is attached on *every* path, including rehydration, so
         errors and terminal outcomes are always captured. But
-        ``conversation_started`` is emitted only for a genuinely new
+        ``conversation_created`` is emitted only for a genuinely new
         conversation: ``_start_event_service`` also runs when an idle
         conversation is lazily reloaded and when RUNNING conversations are
-        recovered after a restart, and counting those as starts would inflate
+        recovered after a restart, and counting those as creations would inflate
         the metric on every server bounce.
 
         Deliberately total: telemetry must never be able to fail conversation

openhands-agent-server/openhands/agent_server/telemetry/factory.py

@@ -137,11 +137,13 @@ def build(
         *,
         user_id: str | None = None,
         occurred_at: datetime | None = None,
+        insert_id: str | None = None,
     ) -> DiagnosticEvent:
         return DiagnosticEvent(
             event_name=event_name,
             schema_version=TELEMETRY_SCHEMA_VERSION,
             occurred_at=occurred_at or utc_now(),
+            insert_id=insert_id,
             distinct_id=self.distinct_id(user_id),
             runtime=self._runtime,
             properties=properties,

openhands-agent-server/openhands/agent_server/telemetry/models.py

@@ -67,6 +67,7 @@ class EventName(StrEnum):
     SERVER_STARTED = "agent_server.server_started"
     SERVER_STOPPED = "agent_server.server_stopped"
     CONVERSATION_STARTED = "agent_server.conversation_started"
+    CONVERSATION_CREATED = "conversation_created"
     CONVERSATION_FINISHED = "agent_server.conversation_finished"
     CONVERSATION_FAILED = "agent_server.conversation_failed"
     CONVERSATION_ERROR = "agent_server.conversation_error"
@@ -245,6 +246,7 @@ class DiagnosticEvent(BaseModel):
     event_name: EventName
     schema_version: int = TELEMETRY_SCHEMA_VERSION
     occurred_at: datetime
+    insert_id: SafeToken | None = None
 
     distinct_id: Annotated[str, StringConstraints(min_length=1, max_length=256)]
     """Correlation identity, passed through verbatim.
@@ -271,6 +273,8 @@ def to_payload(self) -> dict[str, object]:
             **self.runtime.model_dump(mode="json"),
             **self.properties.model_dump(mode="json", exclude={"kind"}),
         }
+        if self.insert_id is not None:
+            payload["$insert_id"] = self.insert_id
         return payload
 
 
@@ -287,6 +291,7 @@ def to_payload(self) -> dict[str, object]:
         "platform",
         "deferred_init",
         "source",
+        "$insert_id",
         "conversation_ref",
         "llm_model_family",
         "agent_kind",

openhands-agent-server/openhands/agent_server/telemetry/subscriber.py

@@ -135,7 +135,7 @@ def _handle_state_update(self, event: ConversationStateUpdateEvent) -> None:
             self._emit_terminal(status)
 
     def emit_started(self) -> None:
-        """Emit ``conversation_started``. Called once, at registration."""
+        """Emit canonical ``conversation_created`` once at registration."""
         try:
             properties = m.ConversationStartedProperties(
                 conversation_ref=self.context.conversation_ref,
@@ -149,13 +149,14 @@ def emit_started(self) -> None:
             )
             self.sink.emit(
                 self.factory.build(
-                    m.EventName.CONVERSATION_STARTED,
+                    m.EventName.CONVERSATION_CREATED,
                     properties,
                     user_id=self.context.user_id,
+                    insert_id=(f"conversation_created:{self.context.conversation_ref}"),
                 )
             )
         except Exception:
-            logger.debug("Telemetry failed to emit conversation_started", exc_info=True)
+            logger.debug("Telemetry failed to emit conversation_created", exc_info=True)
 
     def _emit_terminal(self, status: str) -> None:
         if self._terminal_emitted:
@@ -287,7 +288,7 @@ async def close(self) -> None:
         conversation. Emitting unconditionally here produced a
         ``conversation_finished`` — carrying a non-terminal
         ``terminal_status`` like ``paused`` — for a conversation that did
-        nothing this session, with no matching ``conversation_started``, and
+        nothing this session, with no matching ``conversation_created``, and
         again on every view-then-restart cycle for the same
         ``conversation_ref``.
 

tests/agent_server/telemetry/test_telemetry_disabled_by_default.py

@@ -106,7 +106,7 @@ async def test_conversation_service_reads_the_live_sink_not_a_captured_one(
     captured at construction, every conversation would see the pre-init NoOp and
     emit nothing regardless of consent. Build the service while the sink is still a
     NoOp, enable telemetry afterwards, and assert a new conversation still attaches
-    the subscriber and emits ``conversation_started``.
+    the subscriber and emits ``conversation_created``.
     """
     from uuid import uuid4
 
@@ -178,7 +178,7 @@ async def subscribe_to_events(self, subscriber):
     )
 
     assert len(event_service.subscribers) == 1
-    assert m.EventName.CONVERSATION_STARTED in sink.events
+    assert m.EventName.CONVERSATION_CREATED in sink.events
 
 
 def test_app_exposes_a_sink_on_state_after_startup(temp_persistence_dir):

tests/agent_server/telemetry/test_telemetry_end_to_end.py

@@ -131,7 +131,7 @@ async def test_opted_in_session_emits_sanitized_lifecycle_and_error_events():
     await asyncio.wait_for(sink.aclose(), timeout=10)
 
     names = [p["event"] for p in exporter.payloads]
-    assert m.EventName.CONVERSATION_STARTED in names
+    assert m.EventName.CONVERSATION_CREATED in names
     assert m.EventName.CONVERSATION_ERROR in names
     assert m.EventName.CONVERSATION_FINISHED in names
 
@@ -173,7 +173,10 @@ async def test_opted_in_session_reports_useful_diagnostics():
 
     by_name = {p["event"]: p["properties"] for p in exporter.payloads}
 
-    started = by_name[m.EventName.CONVERSATION_STARTED]
+    started = by_name[m.EventName.CONVERSATION_CREATED]
+    assert (
+        started["$insert_id"] == f"conversation_created:{started['conversation_ref']}"
+    )
     assert started["llm_model_family"] == "anthropic"
     assert started["tool_count"] == 4
 

tests/agent_server/telemetry/test_telemetry_schema.py

@@ -76,6 +76,7 @@ def test_property_names_match_the_declared_allowlist():
     actual: set[str] = {"schema_version"}
     for model in PROPERTY_MODELS:
         actual.update(n for n in model.model_fields if n != "kind")
+    actual.add("$insert_id")
 
     assert actual == set(m.EXPECTED_PROPERTY_NAMES), (
         "Diagnostic property set changed. Update EXPECTED_PROPERTY_NAMES "

tests/agent_server/telemetry/test_telemetry_subscriber.py

@@ -79,20 +79,20 @@ def make_subscriber(sink, factory, user_id: str | None = "user-1"):
 # ── lifecycle ─────────────────────────────────────────────────────────────
 
 
-async def test_emits_exactly_one_started_event(factory):
+async def test_emits_exactly_one_created_event(factory):
     sink = CollectingSink()
     sub = make_subscriber(sink, factory)
 
     sub.emit_started()
-    assert sink.names == [m.EventName.CONVERSATION_STARTED]
+    assert sink.names == [m.EventName.CONVERSATION_CREATED]
 
 
 def test_started_is_only_emitted_for_genuinely_new_conversations():
     """Regression: ``_start_event_service`` also runs on rehydration.
 
     It is called when an idle conversation is lazily reloaded and when RUNNING
     conversations are recovered after a restart. Emitting
-    ``conversation_started`` from all of those would inflate the metric on
+    ``conversation_created`` from all of those would inflate the metric on
     every server bounce, so the flag must default to *not* emitting.
     """
     import inspect
@@ -103,7 +103,7 @@ def test_started_is_only_emitted_for_genuinely_new_conversations():
     param = sig.parameters["is_new_conversation"]
 
     assert param.default is False, (
-        "_start_event_service must default to NOT emitting conversation_started; "
+        "_start_event_service must default to NOT emitting conversation_created; "
         "the hydration path relies on that default"
     )
     assert param.kind is inspect.Parameter.KEYWORD_ONLY
@@ -156,7 +156,7 @@ async def test_close_is_silent_when_no_run_was_observed(factory):
 
     The subscriber attaches on every _start_event_service path, including the
     lazy attach when a user merely views an old conversation. Emitting on close
-    produced a conversation_finished with no matching conversation_started,
+    produced a conversation_finished with no matching conversation_created,
     repeated on every view-then-restart cycle for the same conversation_ref.
     """
     sink = CollectingSink()
@@ -397,7 +397,7 @@ async def test_disabled_sink_short_circuits_before_building_events(factory):
 
     sub.emit_started()
     # emit() itself is a no-op on a disabled sink; nothing is recorded.
-    assert sink.events == [] or sink.names == [m.EventName.CONVERSATION_STARTED]
+    assert sink.events == [] or sink.names == [m.EventName.CONVERSATION_CREATED]
 
 
 # ── identity ──────────────────────────────────────────────────────────────