Simplify TimeInput type to just use datetime

Copilot · nikhilNava · Copilot · commit 8354142efc03 · 2026-02-24T19:28:03.000Z
Based on user feedback, simplify the start_time/end_time parameters to only accept
datetime objects instead of the complex TimeInput union type. This is more Pythonic
and consistent with existing code in the LangChain tracer.

Changes:
- Remove TimeInput type alias
- Change start_time/end_time parameters to datetime | None
- Add simple _datetime_to_ns() conversion method
- Remove unused _time_input_to_ns() and _time_input_to_ms() methods
- Update all scope subclasses to use datetime | None
- Simplify tests to only use datetime

Co-authored-by: nikhilNava &lt;211831449+nikhilNava@users.noreply.github.com&gt;
diff --git a/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/__init__.py b/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/__init__.py
@@ -24,7 +24,7 @@
 from .invoke_agent_details import InvokeAgentDetails
 from .invoke_agent_scope import InvokeAgentScope
 from .middleware.baggage_builder import BaggageBuilder
-from .opentelemetry_scope import OpenTelemetryScope, TimeInput
+from .opentelemetry_scope import OpenTelemetryScope
 from .request import Request
 from .source_metadata import SourceMetadata
 from .tenant_details import TenantDetails
@@ -47,8 +47,6 @@
     "SpanProcessor",
     # Base scope class
     "OpenTelemetryScope",
-    # Type aliases
-    "TimeInput",
     # Specific scope classes
     "ExecuteToolScope",
     "InvokeAgentScope",
diff --git a/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/execute_tool_scope.py b/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/execute_tool_scope.py
@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+from datetime import datetime
+
 from .agent_details import AgentDetails
 from .constants import (
     EXECUTE_TOOL_OPERATION_NAME,
@@ -15,7 +17,7 @@
     SERVER_ADDRESS_KEY,
     SERVER_PORT_KEY,
 )
-from .opentelemetry_scope import OpenTelemetryScope, TimeInput
+from .opentelemetry_scope import OpenTelemetryScope
 from .request import Request
 from .tenant_details import TenantDetails
 from .tool_call_details import ToolCallDetails
@@ -31,8 +33,8 @@ def start(
         tenant_details: TenantDetails,
         request: Request | None = None,
         parent_id: str | None = None,
-        start_time: TimeInput = None,
-        end_time: TimeInput = None,
+        start_time: datetime | None = None,
+        end_time: datetime | None = None,
     ) -> "ExecuteToolScope":
         """Creates and starts a new scope for tool execution tracing.
 
@@ -43,11 +45,10 @@ def start(
             request: Optional request details for additional context
             parent_id: Optional parent Activity ID used to link this span to an upstream
                 operation
-            start_time: Optional explicit start time. Accepts int (nanoseconds since epoch),
-                float (seconds since epoch), tuple[int, int] (HrTime), or datetime. Useful when
+            start_time: Optional explicit start time as a datetime object. Useful when
                 recording a tool call after execution has already completed.
-            end_time: Optional explicit end time in the same formats as start_time. When
-                provided, the span will use this timestamp when disposed instead of the
+            end_time: Optional explicit end time as a datetime object. When provided,
+                the span will use this timestamp when disposed instead of the
                 current wall-clock time.
 
         Returns:
@@ -64,8 +65,8 @@ def __init__(
         tenant_details: TenantDetails,
         request: Request | None = None,
         parent_id: str | None = None,
-        start_time: TimeInput = None,
-        end_time: TimeInput = None,
+        start_time: datetime | None = None,
+        end_time: datetime | None = None,
     ):
         """Initialize the tool execution scope.
 
@@ -76,11 +77,10 @@ def __init__(
             request: Optional request details for additional context
             parent_id: Optional parent Activity ID used to link this span to an upstream
                 operation
-            start_time: Optional explicit start time. Accepts int (nanoseconds since epoch),
-                float (seconds since epoch), tuple[int, int] (HrTime), or datetime. Useful when
+            start_time: Optional explicit start time as a datetime object. Useful when
                 recording a tool call after execution has already completed.
-            end_time: Optional explicit end time in the same formats as start_time. When
-                provided, the span will use this timestamp when disposed instead of the
+            end_time: Optional explicit end time as a datetime object. When provided,
+                the span will use this timestamp when disposed instead of the
                 current wall-clock time.
         """
         super().__init__(
diff --git a/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/inference_scope.py b/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/inference_scope.py
@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+from datetime import datetime
 from typing import List
 
 from .agent_details import AgentDetails
@@ -19,7 +20,7 @@
     GEN_AI_USAGE_OUTPUT_TOKENS_KEY,
 )
 from .inference_call_details import InferenceCallDetails
-from .opentelemetry_scope import OpenTelemetryScope, TimeInput
+from .opentelemetry_scope import OpenTelemetryScope
 from .request import Request
 from .tenant_details import TenantDetails
 from .utils import safe_json_dumps
@@ -35,8 +36,8 @@ def start(
         tenant_details: TenantDetails,
         request: Request | None = None,
         parent_id: str | None = None,
-        start_time: TimeInput = None,
-        end_time: TimeInput = None,
+        start_time: datetime | None = None,
+        end_time: datetime | None = None,
     ) -> "InferenceScope":
         """Creates and starts a new scope for inference tracing.
 
@@ -47,9 +48,8 @@ def start(
             request: Optional request details for additional context
             parent_id: Optional parent Activity ID used to link this span to an upstream
                 operation
-            start_time: Optional explicit start time. Accepts int (nanoseconds since epoch),
-                float (seconds since epoch), tuple[int, int] (HrTime), or datetime.
-            end_time: Optional explicit end time in the same formats as start_time.
+            start_time: Optional explicit start time as a datetime object.
+            end_time: Optional explicit end time as a datetime object.
 
         Returns:
             A new InferenceScope instance
@@ -65,8 +65,8 @@ def __init__(
         tenant_details: TenantDetails,
         request: Request | None = None,
         parent_id: str | None = None,
-        start_time: TimeInput = None,
-        end_time: TimeInput = None,
+        start_time: datetime | None = None,
+        end_time: datetime | None = None,
     ):
         """Initialize the inference scope.
 
@@ -77,9 +77,8 @@ def __init__(
             request: Optional request details for additional context
             parent_id: Optional parent Activity ID used to link this span to an upstream
                 operation
-            start_time: Optional explicit start time. Accepts int (nanoseconds since epoch),
-                float (seconds since epoch), tuple[int, int] (HrTime), or datetime.
-            end_time: Optional explicit end time in the same formats as start_time.
+            start_time: Optional explicit start time as a datetime object.
+            end_time: Optional explicit end time as a datetime object.
         """
 
         super().__init__(
diff --git a/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/invoke_agent_scope.py b/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/invoke_agent_scope.py
@@ -4,6 +4,7 @@
 # Invoke agent scope for tracing agent invocation.
 
 import logging
+from datetime import datetime
 
 from .agent_details import AgentDetails
 from .constants import (
@@ -32,7 +33,7 @@
 )
 from .invoke_agent_details import InvokeAgentDetails
 from .models.caller_details import CallerDetails
-from .opentelemetry_scope import OpenTelemetryScope, TimeInput
+from .opentelemetry_scope import OpenTelemetryScope
 from .request import Request
 from .tenant_details import TenantDetails
 from .utils import safe_json_dumps, validate_and_normalize_ip
@@ -50,8 +51,8 @@ def start(
         request: Request | None = None,
         caller_agent_details: AgentDetails | None = None,
         caller_details: CallerDetails | None = None,
-        start_time: TimeInput = None,
-        end_time: TimeInput = None,
+        start_time: datetime | None = None,
+        end_time: datetime | None = None,
     ) -> "InvokeAgentScope":
         """Create and start a new scope for agent invocation tracing.
 
@@ -62,9 +63,8 @@ def start(
             request: Optional request details for additional context
             caller_agent_details: Optional details of the caller agent
             caller_details: Optional details of the non-agentic caller
-            start_time: Optional explicit start time. Accepts int (nanoseconds since epoch),
-                float (seconds since epoch), tuple[int, int] (HrTime), or datetime.
-            end_time: Optional explicit end time in the same formats as start_time.
+            start_time: Optional explicit start time as a datetime object.
+            end_time: Optional explicit end time as a datetime object.
 
         Returns:
             A new InvokeAgentScope instance
@@ -86,8 +86,8 @@ def __init__(
         request: Request | None = None,
         caller_agent_details: AgentDetails | None = None,
         caller_details: CallerDetails | None = None,
-        start_time: TimeInput = None,
-        end_time: TimeInput = None,
+        start_time: datetime | None = None,
+        end_time: datetime | None = None,
     ):
         """Initialize the agent invocation scope.
 
@@ -97,9 +97,8 @@ def __init__(
             request: Optional request details for additional context
             caller_agent_details: Optional details of the caller agent
             caller_details: Optional details of the non-agentic caller
-            start_time: Optional explicit start time. Accepts int (nanoseconds since epoch),
-                float (seconds since epoch), tuple[int, int] (HrTime), or datetime.
-            end_time: Optional explicit end time in the same formats as start_time.
+            start_time: Optional explicit start time as a datetime object.
+            end_time: Optional explicit end time as a datetime object.
         """
         activity_name = INVOKE_AGENT_OPERATION_NAME
         if invoke_agent_details.details.agent_name:
diff --git a/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/opentelemetry_scope.py b/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/opentelemetry_scope.py
@@ -48,12 +48,6 @@
 # Create logger for this module - inherits from 'microsoft_agents_a365.observability.core'
 logger = logging.getLogger(__name__)
 
-# TimeInput is a type alias for types that can be used as span timestamps.
-# OpenTelemetry Python SDK accepts int (nanoseconds since epoch), float (seconds since epoch),
-# or a tuple of (seconds, nanoseconds) as HrTime.
-# We extend this to also accept datetime objects for convenience.
-TimeInput = int | float | tuple[int, int] | datetime | None
-
 
 class OpenTelemetryScope:
     """Base class for OpenTelemetry tracing scopes in the SDK."""
@@ -78,6 +72,20 @@ def _is_telemetry_enabled(cls) -> bool:
         enable_observability = os.getenv(ENABLE_A365_OBSERVABILITY, "").lower()
         return (env_value or enable_observability) in ("true", "1", "yes", "on")
 
+    @staticmethod
+    def _datetime_to_ns(dt: datetime | None) -> int | None:
+        """Convert a datetime to nanoseconds since epoch.
+
+        Args:
+            dt: Python datetime object, or None
+
+        Returns:
+            Nanoseconds since epoch, or None if input is None
+        """
+        if dt is None:
+            return None
+        return int(dt.timestamp() * 1_000_000_000)
+
     def __init__(
         self,
         kind: str,
@@ -86,8 +94,8 @@ def __init__(
         agent_details: "AgentDetails | None" = None,
         tenant_details: "TenantDetails | None" = None,
         parent_id: str | None = None,
-        start_time: TimeInput = None,
-        end_time: TimeInput = None,
+        start_time: datetime | None = None,
+        end_time: datetime | None = None,
     ):
         """Initialize the OpenTelemetry scope.
 
@@ -99,19 +107,15 @@ def __init__(
             tenant_details: Optional tenant details
             parent_id: Optional parent Activity ID used to link this span to an upstream
                 operation
-            start_time: Optional explicit start time. Can be:
-                - int: nanoseconds since epoch
-                - float: seconds since epoch
-                - tuple[int, int]: HrTime as (seconds, nanoseconds)
-                - datetime: Python datetime object
+            start_time: Optional explicit start time as a datetime object.
                 Useful when recording an operation after it has already completed.
-            end_time: Optional explicit end time in the same format as start_time.
+            end_time: Optional explicit end time as a datetime object.
                 When provided, the span will use this timestamp when disposed
                 instead of the current wall-clock time.
         """
         self._span: Span | None = None
-        self._custom_start_time: TimeInput = start_time
-        self._custom_end_time: TimeInput = end_time
+        self._custom_start_time: datetime | None = start_time
+        self._custom_end_time: datetime | None = end_time
         self._has_ended = False
         self._error_type: str | None = None
         self._exception: Exception | None = None
@@ -138,7 +142,7 @@ def __init__(
             span_context = parent_context if parent_context else context.get_current()
 
             # Convert custom start time to OTel-compatible format (nanoseconds since epoch)
-            otel_start_time = self._time_input_to_ns(start_time) if start_time else None
+            otel_start_time = self._datetime_to_ns(start_time)
 
             self._span = tracer.start_span(
                 activity_name,
@@ -256,68 +260,15 @@ def record_attributes(self, attributes: dict[str, Any] | list[tuple[str, Any]])
             if key and key.strip():
                 self._span.set_attribute(key, value)
 
-    @staticmethod
-    def _time_input_to_ns(t: TimeInput) -> int | None:
-        """Convert a TimeInput value to nanoseconds since epoch.
-
-        OpenTelemetry Python SDK accepts int (nanoseconds since epoch) for span
-        start_time and end_time parameters.
-
-        Args:
-            t: TimeInput value which can be:
-                - int: nanoseconds since epoch (returned as-is)
-                - float: seconds since epoch
-                - tuple[int, int]: HrTime as (seconds, nanoseconds)
-                - datetime: Python datetime object
-
-        Returns:
-            Nanoseconds since epoch, or None if input is None
-        """
-        if t is None:
-            return None
-        if isinstance(t, int):
-            # Assume nanoseconds if it's an integer
-            return t
-        if isinstance(t, float):
-            # Convert seconds to nanoseconds
-            return int(t * 1_000_000_000)
-        if isinstance(t, tuple) and len(t) == 2:
-            # HrTime: (seconds, nanoseconds)
-            seconds, nanos = t
-            return seconds * 1_000_000_000 + nanos
-        if isinstance(t, datetime):
-            return int(t.timestamp() * 1_000_000_000)
-        logger.warning(
-            f"_time_input_to_ns received unexpected TimeInput "
-            f"(type={type(t).__name__}); returning None"
-        )
-        return None
-
-    @staticmethod
-    def _time_input_to_ms(t: TimeInput) -> float | None:
-        """Convert a TimeInput value to milliseconds since epoch.
-
-        Used for duration calculations.
-
-        Args:
-            t: TimeInput value (see _time_input_to_ns for accepted types)
-
-        Returns:
-            Milliseconds since epoch, or None if input is None
-        """
-        ns = OpenTelemetryScope._time_input_to_ns(t)
-        return ns / 1_000_000 if ns is not None else None
-
-    def set_end_time(self, end_time: TimeInput) -> None:
+    def set_end_time(self, end_time: datetime) -> None:
         """Set a custom end time for the scope.
 
         When set, dispose() will pass this value to span.end() instead of using
         the current wall-clock time. This is useful when the actual end time of
         the operation is known before the scope is disposed.
 
         Args:
-            end_time: The end time as nanoseconds since epoch, seconds since epoch,
-                     HrTime tuple (seconds, nanoseconds), or datetime object.
+            end_time: The end time as a datetime object.
         """
         self._custom_end_time = end_time
 
@@ -329,7 +280,7 @@ def _end(self) -> None:
             logger.info(f"Span ended: '{self._span.name}' ({span_id})")
 
             # Convert custom end time to OTel-compatible format (nanoseconds since epoch)
-            otel_end_time = self._time_input_to_ns(self._custom_end_time)
+            otel_end_time = self._datetime_to_ns(self._custom_end_time)
             if otel_end_time is not None:
                 self._span.end(end_time=otel_end_time)
             else:
diff --git a/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/spans_scopes/output_scope.py b/libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/spans_scopes/output_scope.py
diff --git a/tests/observability/core/test_custom_start_end_time.py b/tests/observability/core/test_custom_start_end_time.py
