feat(sensor): add opt-in GenAI semconv attributes to OTel session logs

Sensor/README.md

@@ -420,6 +420,36 @@ no redaction or field projection, so prompts, responses, tool arguments, tool
 results, usernames, hostnames, and local paths can be transmitted. Any
 normalization already performed by a source parser still applies.
 
+Set `"gen_ai_attributes": true` to also add OpenTelemetry GenAI semantic-convention
+attributes to `adr.agent.session` records. The option is off by default because the
+GenAI conventions are still in development status. The body and all `adr.*`
+attributes, including `adr.session.id`, are unchanged. Enabling it is a destination
+change and resends sessions once.
+
+`gen_ai.agent.name` is the source (same value as `adr.source`).
+`gen_ai.conversation.id` is the harness's own session ID, so ADR sessions can be
+joined with the telemetry that harness reports itself. Each parser builds
+`adr.session.id` by prefixing that ID, and the exporter maps it back as follows:
+
+| source | ADR `session_id` | Native ID the parser reads | `gen_ai.conversation.id` |
+|---|---|---|---|
+| `antigravity` | `antigravity_<id>` | Conversation directory name | `<id>` |
+| `claude` | `claude_<id>`, subagents `claude_<id>_agent_<agent>` | JSONL `sessionId` | `<id>`; subagents get their parent's `sessionId`, as in Claude Code's own telemetry |
+| `claude_desktop` | `claude_desktop_[dispatch_]<uuid>` | Metadata `cliSessionId` (kept as `session_context.cli_session_id`) | `cliSessionId`; omitted when absent |
+| `cline` | `cline_<id>` | Task directory name (Cline's task ID) | `<id>` |
+| `codex` | `codex_<id>` | `session_meta` `payload.id` | `<id>` |
+| `copilot` | `copilot_<id>` | `session.start`/`session.resume` `sessionId`, else the session-state directory name | `<id>` |
+| `cursor` | `cursor_<id>` | Composer ID from `composerData:`/`bubbleId:` keys | `<id>` |
+| `dsh` | `dsh_<id>` | Session header `id` | `<id>` |
+| `gemini` | `gemini_<id>` | `sessionId`; subagents keep their own | `<id>` |
+| `opencode` | `opencode_<id>` | Session `id`, whose `ses_` prefix the parser strips | `ses_<id>` |
+| `warp` | `warp_<id>` | `agent_conversations.conversation_id` | `<id>` |
+
+`gen_ai.conversation.id` is omitted when it is unavailable: for an unlisted source,
+a `session_id` without the expected prefix, a Claude subagent without a parent
+session, or a Claude Desktop session without `cliSessionId`. It is never derived
+from `adr.session.id`. New parsers need a row in this table.
+
 System-configuration records are sent as `adr.system.configuration` logs on each
 run. Sensor health logs are also sent on every run, even when all session snapshots
 are already acknowledged. With `--save-sessions`, successful session delivery is tracked independently

Sensor/adr_sensor/exporters/config.py

@@ -15,6 +15,7 @@
     "timeout_seconds",
     "flush_timeout_seconds",
     "certificate_file",
+    "gen_ai_attributes",
 }
 
 
@@ -32,6 +33,7 @@ class OpenTelemetryConfig:
     timeout_seconds: float = 10.0
     flush_timeout_seconds: float = 30.0
     certificate_file: Optional[str] = None
+    gen_ai_attributes: bool = False
 
 
 def load_opentelemetry_config(path: Path) -> OpenTelemetryConfig:
@@ -84,13 +86,18 @@ def load_opentelemetry_config(path: Path) -> OpenTelemetryConfig:
         if not Path(certificate_file).is_file():
             raise OpenTelemetryConfigError(f"OpenTelemetry certificate file does not exist: {certificate_file}")
 
+    gen_ai_attributes = document.get("gen_ai_attributes", False)
+    if not isinstance(gen_ai_attributes, bool):
+        raise OpenTelemetryConfigError("OpenTelemetry gen_ai_attributes must be a boolean")
+
     return OpenTelemetryConfig(
         endpoint=endpoint,
         service_name=service_name,
         headers=dict(headers_value),
         timeout_seconds=timeout_seconds,
         flush_timeout_seconds=flush_timeout_seconds,
         certificate_file=certificate_file,
+        gen_ai_attributes=gen_ai_attributes,
     )
 
 

Sensor/adr_sensor/exporters/delivery_checkpoint.py

@@ -34,8 +34,12 @@ class DeliveryCheckpoint:
 
     def __init__(self, output_dir: Path, config: OpenTelemetryConfig):
         _validate_credential_provider()
+        config_fields = asdict(config)
+        if not config.gen_ai_attributes:
+            # Keep checkpoints from before this option existed valid when it is off.
+            del config_fields["gen_ai_attributes"]
         destination = {
-            "config": asdict(config),
+            "config": config_fields,
             "schema_version": SCHEMA_VERSION,
             "checkpoint_version": 1,
         }

Sensor/adr_sensor/exporters/opentelemetry.py

@@ -4,7 +4,7 @@
 import threading
 import time
 from datetime import datetime, timezone
-from typing import Any, List, Optional, Tuple
+from typing import Any, Callable, Dict, List, Optional, Tuple
 
 from .. import host_identity
 from ..diagnostics import sanitize_health_record
@@ -80,6 +80,7 @@ def __init__(
         self._info_severity = severity_number_cls.INFO
         self._warning_severity = severity_number_cls.WARN
         self._flush_timeout_millis = int(config.flush_timeout_seconds * 1000)
+        self._gen_ai_attributes = config.gen_ai_attributes
         self._closed = False
         self._submitted_count = 0
         self._flushed_count = 0
@@ -103,6 +104,7 @@ def export(
                     "adr.source": entry.source,
                     "adr.session.id": entry.session_id,
                     **({"adr.model": entry.model} if entry.model else {}),
+                    **(_gen_ai_attributes(entry) if self._gen_ai_attributes else {}),
                 },
             )
 
@@ -204,6 +206,71 @@ def _emit(
                 self._flush()
 
 
+def _gen_ai_attributes(entry: AgentEvent) -> dict:
+    """Map session identity onto OpenTelemetry GenAI semantic-convention attributes."""
+    attributes = {"gen_ai.agent.name": entry.source}
+    conversation_id = _conversation_id(entry)
+    if conversation_id:
+        attributes["gen_ai.conversation.id"] = conversation_id
+    return attributes
+
+
+def _conversation_id(entry: AgentEvent) -> Optional[str]:
+    """Return the harness's own session ID, or None when it cannot be recovered.
+
+    Never falls back to ``entry.session_id``: ``adr.session.id`` already carries it.
+    """
+    native_id = _CONVERSATION_ID_BY_SOURCE.get(entry.source)
+    return native_id(entry) if native_id else None
+
+
+def _without_prefix(value: Any, prefix: str) -> Optional[str]:
+    if isinstance(value, str) and value.startswith(prefix) and len(value) > len(prefix):
+        return value[len(prefix) :]
+    return None
+
+
+def _prefixed(prefix: str) -> Callable[[AgentEvent], Optional[str]]:
+    """Undo a parser's ``f"{prefix}{native_id}"`` session ID."""
+    return lambda entry: _without_prefix(entry.session_id, prefix)
+
+
+def _claude_conversation_id(entry: AgentEvent) -> Optional[str]:
+    context = entry.session_context or {}
+    if context.get("agent_id"):
+        # Subagent transcripts carry their parent's sessionId, which Claude Code also reports.
+        return _without_prefix(context.get("parent_session_id"), "claude_")
+    return _without_prefix(entry.session_id, "claude_")
+
+
+def _claude_desktop_conversation_id(entry: AgentEvent) -> Optional[str]:
+    # The metadata sessionId prefix is stripped by the parser; cliSessionId is kept verbatim.
+    cli_session_id = (entry.session_context or {}).get("cli_session_id")
+    return cli_session_id if isinstance(cli_session_id, str) and cli_session_id else None
+
+
+def _opencode_conversation_id(entry: AgentEvent) -> Optional[str]:
+    # The parser strips opencode's own "ses_" prefix, which every opencode session ID has.
+    native_id = _without_prefix(entry.session_id, "opencode_")
+    return f"ses_{native_id}" if native_id else None
+
+
+# gen_ai.conversation.id for each AgentEvent.source; see "OpenTelemetry Logs Export" in Sensor/README.md.
+_CONVERSATION_ID_BY_SOURCE: Dict[str, Callable[[AgentEvent], Optional[str]]] = {
+    "antigravity": _prefixed("antigravity_"),
+    "claude": _claude_conversation_id,
+    "claude_desktop": _claude_desktop_conversation_id,
+    "cline": _prefixed("cline_"),
+    "codex": _prefixed("codex_"),
+    "copilot": _prefixed("copilot_"),
+    "cursor": _prefixed("cursor_"),
+    "dsh": _prefixed("dsh_"),
+    "gemini": _prefixed("gemini_"),
+    "opencode": _opencode_conversation_id,
+    "warp": _prefixed("warp_"),
+}
+
+
 def _datetime_to_unix_nanos(value: datetime) -> int:
     if value.tzinfo is None or value.utcoffset() is None:
         value = value.replace(tzinfo=timezone.utc)

Sensor/tests/test_delivery_checkpoint.py

@@ -92,6 +92,26 @@ def test_destination_or_authentication_change_resends_sessions(tmp_path, config,
     assert changed.pending_entries([event]) == [event]
 
 
+def test_default_gen_ai_attributes_keep_existing_checkpoint(tmp_path, config, monkeypatch):
+    for prefix in ("OTEL_EXPORTER_OTLP_", "OTEL_EXPORTER_OTLP_LOGS_"):
+        for setting in ("CLIENT_CERTIFICATE", "CLIENT_KEY"):
+            monkeypatch.delenv(f"{prefix}{setting}", raising=False)
+
+    # Checkpoint name written for this configuration before gen_ai_attributes existed.
+    expected = ".adr-otel-delivery.5fae3b25f180fb0d10b1da8a0e95e774e8f1cb93f7db68fc9440d99c39858fa6.json"
+    assert DeliveryCheckpoint(tmp_path, config).path.name == expected
+
+
+def test_enabling_gen_ai_attributes_resends_sessions(tmp_path, config, event):
+    checkpoint = DeliveryCheckpoint(tmp_path, config)
+    checkpoint.pending_entries([event])
+    checkpoint.commit()
+    enabled = DeliveryCheckpoint(tmp_path, replace(config, gen_ai_attributes=True))
+
+    assert enabled.path != checkpoint.path
+    assert enabled.pending_entries([event]) == [event]
+
+
 def test_environment_authentication_change_resends_sessions(tmp_path, event, monkeypatch):
     config = OpenTelemetryConfig(endpoint="https://collector.example.invalid/v1/logs")
     monkeypatch.setenv("OTEL_EXPORTER_OTLP_LOGS_HEADERS", "Authorization=first-synthetic-token")

Sensor/tests/test_exporter_config.py

@@ -25,6 +25,7 @@ def test_load_opentelemetry_config_defaults(tmp_path):
     assert config.timeout_seconds == 10.0
     assert config.flush_timeout_seconds == 30.0
     assert config.certificate_file is None
+    assert config.gen_ai_attributes is False
 
 
 def test_load_opentelemetry_config_all_fields(tmp_path):
@@ -40,6 +41,7 @@ def test_load_opentelemetry_config_all_fields(tmp_path):
                 "timeout_seconds": 5,
                 "flush_timeout_seconds": 12.5,
                 "certificate_file": certificate.name,
+                "gen_ai_attributes": True,
             },
         )
     )
@@ -49,6 +51,7 @@ def test_load_opentelemetry_config_all_fields(tmp_path):
     assert config.timeout_seconds == 5.0
     assert config.flush_timeout_seconds == 12.5
     assert config.certificate_file == str(certificate.resolve())
+    assert config.gen_ai_attributes is True
 
 
 @pytest.mark.parametrize(
@@ -60,6 +63,8 @@ def test_load_opentelemetry_config_all_fields(tmp_path):
         ({"endpoint": "http://localhost:4318/v1/logs", "headers": []}, "headers must be an object"),
         ({"endpoint": "http://localhost:4318/v1/logs", "timeout_seconds": 0}, "positive number"),
         ({"endpoint": "http://localhost:4318/v1/logs", "timeout_seconds": float("nan")}, "positive number"),
+        ({"endpoint": "http://localhost:4318/v1/logs", "gen_ai_attributes": "true"}, "must be a boolean"),
+        ({"endpoint": "http://localhost:4318/v1/logs", "gen_ai_attributes": 1}, "must be a boolean"),
         ({"endpoint": "http://localhost:4318/v1/logs", "unexpected": True}, "unknown"),
     ],
 )