fix: address comment

docs/contributing/go/spanmapper.md

@@ -17,7 +17,7 @@ Condition: `model`, `llm.input_messages` (OpenInference sets `llm.model_name` fr
 | `gen_ai.response.model`, last source for the request model | vLLM, which writes only the response model |
 | `ai.model.id`, `ai.model.provider`, `ai.response.model`, `ai.response.finishReason`, `ai.prompt.messages`, `ai.response.text`, `ai.response.toolCalls`, `ai.usage.promptTokens`, `ai.usage.completionTokens` | Vercel AI SDK v4, Mastra |
 | `ai.usage.inputTokens`, `ai.usage.outputTokens`, `ai.usage.cachedInputTokens`, `ai.usage.reasoningTokens` | Vercel AI SDK v5 |
-| `langfuse.observation.model.name`, `langfuse.session.id` | Langfuse SDK v3 |
+| `langfuse.observation.model.name`, `langfuse.observation.input`, `langfuse.observation.output`, `langfuse.observation.type`, `langfuse.session.id` | Langfuse SDK v3 (Python) and v4 (JS), LiteLLM `langfuse_otel` callback |
 | `session.id` | OpenInference, Langfuse, Claude Code |
 | `gen_ai.client.session_id` | opentelemetry-hooks runner (Cursor, Copilot, Claude Code hooks) |
 | `model`, `input_tokens`, `output_tokens`, `cache_read_tokens`, `cache_creation_tokens` | Claude Code `claude_code.llm_request` spans ([docs](https://code.claude.com/docs/en/monitoring-usage#span-attributes)) |

pkg/modules/spanmapper/implspanmapper/fs/definitions/gen_ai.llm.json

@@ -1,5 +1,5 @@
 {
-  "version": 4,
+  "version": 5,
   "definition": {
     "name": "gen_ai.llm",
     "condition": {
@@ -134,6 +134,12 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "langfuse.observation.type",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 40
+            },
             {
               "key": "openinference.span.kind",
               "context": "attribute",
@@ -356,6 +362,12 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "langfuse.observation.input",
+              "context": "attribute",
+              "operation": "move",
+              "priority": 40
+            },
             {
               "key": "gen_ai.prompt",
               "context": "attribute",
@@ -382,6 +394,12 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "langfuse.observation.output",
+              "context": "attribute",
+              "operation": "move",
+              "priority": 40
+            },
             {
               "key": "gen_ai.completion",
               "context": "attribute",

pkg/querier/api.go

@@ -140,9 +140,7 @@ func (handler *handler) QueryRawStream(rw http.ResponseWriter, req *http.Request
 	}
 	// create the v5 request param
 	queryRangeRequest := qbtypes.QueryRangeRequest{
-		// Coerce the window to epoch milliseconds up front (End may be 0 for the
-		// open-ended stream, which ToMilliSecs leaves untouched).
-		Start:       qbtypes.ToMilliSecs(start),
+		Start:       start,
 		RequestType: qbtypes.RequestTypeRawStream,
 		CompositeQuery: qbtypes.CompositeQuery{
 			Queries: []qbtypes.QueryEnvelope{

pkg/querier/list_range.go

@@ -11,12 +11,6 @@ func makeBuckets(start, end uint64) []tsRange {
 	startNS := querybuilder.ToNanoSecs(start)
 	endNS := querybuilder.ToNanoSecs(end)
 
-	// A cursor that reaches the window boundary can leave end <= start; the
-	// subtractions below are unsigned and would wrap.
-	if endNS <= startNS {
-		return nil
-	}
-
 	if endNS-startNS <= uint64(hourNanos) {
 		return []tsRange{{fromNS: startNS, toNS: endNS}}
 	}
@@ -26,9 +20,9 @@ func makeBuckets(start, end uint64) []tsRange {
 	curEnd := endNS
 
 	for {
-		curStart := startNS
-		if curEnd-startNS > bucket {
-			curStart = curEnd - bucket
+		curStart := curEnd - bucket
+		if curStart < startNS {
+			curStart = startNS
 		}
 		out = append(out, tsRange{fromNS: curStart, toNS: curEnd})
 

pkg/querier/querier.go

@@ -111,10 +111,11 @@ func New(
 
 func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (*qbtypes.QueryRangeResponse, error) {
 
-	// Callers that build the request in-process never pass through
-	// UnmarshalJSON, so normalise here too.
-	req.Start = qbtypes.ToMilliSecs(req.Start)
-	req.End = qbtypes.ToMilliSecs(req.End)
+	// Coerce the window to epoch milliseconds up front so every downstream
+	// consumer (TimeRange, narrowWindowByTraceID, step interval, etc.) can
+	// safely assume ms regardless of the resolution the caller sent.
+	req.Start = querybuilder.ToMilliSecs(req.Start)
+	req.End = querybuilder.ToMilliSecs(req.End)
 
 	event := &qbtypes.QBEvent{
 		Version:         "v5",
@@ -533,8 +534,10 @@ func familyTemporality(temporalities map[string]metrictypes.Temporality, names [
 
 func (q *querier) QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, client *qbtypes.RawStream) {
 
-	req.Start = qbtypes.ToMilliSecs(req.Start)
-	req.End = qbtypes.ToMilliSecs(req.End)
+	// Coerce the window to epoch milliseconds up front (End may be 0 for the
+	// open-ended stream, which ToMilliSecs leaves untouched).
+	req.Start = querybuilder.ToMilliSecs(req.Start)
+	req.End = querybuilder.ToMilliSecs(req.End)
 
 	event := &qbtypes.QBEvent{
 		Version:         "v5",

pkg/querybuilder/time.go

@@ -33,6 +33,28 @@ func ToNanoSecs(epoch uint64) uint64 {
 	return temp * uint64(math.Pow(10, float64(19-count)))
 }
 
+// ToMilliSecs takes an epoch whose resolution is inferred from its magnitude
+// (s/ms/µs/ns) and returns it in milliseconds. A millisecond epoch for the
+// current era has 13 digits (e.g. ~1.7e12 in 2026), so the value is scaled so
+// its digit-width matches: smaller values (seconds) are scaled up, larger ones
+// (micro/nanoseconds) are scaled down. Zero is returned unchanged.
+func ToMilliSecs(epoch uint64) uint64 {
+	if epoch == 0 {
+		return 0
+	}
+	temp := epoch
+	count := 0
+	for epoch != 0 {
+		epoch /= 10
+		count++
+	}
+	const msDigits = 13
+	if count < msDigits {
+		return temp * uint64(math.Pow(10, float64(msDigits-count)))
+	}
+	return temp / uint64(math.Pow(10, float64(count-msDigits)))
+}
+
 // TODO(srikanthccv): should these be rounded to nearest multiple of 60 instead of 5 if step > 60?
 // That would make graph look nice but "nice" but should be less important than the usefulness.
 func RecommendedStepInterval(start, end uint64) uint64 {

pkg/querybuilder/time_test.go

@@ -60,3 +60,51 @@ func TestToNanoSecs(t *testing.T) {
 		})
 	}
 }
+
+func TestToMilliSecs(t *testing.T) {
+	tests := []struct {
+		name     string
+		epoch    uint64
+		expected uint64
+	}{
+		{
+			name:     "10-digit Unix timestamp (seconds) - 2023-01-01 00:00:00 UTC",
+			epoch:    1672531200,    // seconds
+			expected: 1672531200000, // * 10^3
+		},
+		{
+			name:     "13-digit Unix timestamp (milliseconds) - already ms",
+			epoch:    1672531200000,
+			expected: 1672531200000, // unchanged
+		},
+		{
+			name:     "16-digit Unix timestamp (microseconds)",
+			epoch:    1672531200000000, // microseconds
+			expected: 1672531200000,    // / 10^3
+		},
+		{
+			name:     "19-digit Unix timestamp (nanoseconds)",
+			epoch:    1672531200000000000, // nanoseconds
+			expected: 1672531200000,       // / 10^6
+		},
+		{
+			name:     "Unix epoch start - zero is unchanged",
+			epoch:    0,
+			expected: 0,
+		},
+		{
+			name:     "Recent timestamp in seconds - 2024-05-25 12:00:00 UTC",
+			epoch:    1716638400,
+			expected: 1716638400000,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := ToMilliSecs(tt.epoch)
+			if result != tt.expected {
+				t.Errorf("ToMilliSecs(%d) = %d, want %d", tt.epoch, result, tt.expected)
+			}
+		})
+	}
+}

pkg/types/aiobservabilitytypes/message_format.go

@@ -121,10 +121,13 @@ func (f *formatting) messages() {
 
 // OpenRouter's answer object or an Anthropic response beside the OpenAI chat request it answers is one
 // format, not a mix: an Anthropic request without system and without Anthropic-only blocks reads as an
-// OpenAI chat request.
+// OpenAI chat request. A side kept as generic already warned, and a side read from OpenInference's
+// buckets beside the raw payload of the other is the documented fallback.
 func (f *formatting) warnMixedFormats(inputLabel, outputLabel string) {
 	switch {
 	case inputLabel == "" || outputLabel == "" || outputLabel == inputLabel:
+	case inputLabel == FormatterGeneric || outputLabel == FormatterGeneric:
+	case inputLabel == FormatterOpenInference || outputLabel == FormatterOpenInference:
 	case inputLabel == FormatterOpenAIChat && (outputLabel == FormatterOpenRouter || outputLabel == FormatterAnthropic):
 	default:
 		f.warn("output formatted as %s, input as %s", outputLabel, inputLabel)
@@ -183,11 +186,16 @@ func (f *formatting) side(value any, role Role) (OutputMessages, string) {
 				return f.semconv(v), FormatterSemconv
 			case isAnthropicMessages(v):
 				return f.anthropicMessages(v), FormatterAnthropic
+			case isResponsesItems(v):
+				return foldAssistant(f.responsesItems(v)), FormatterOpenAIResponses
 			case first["role"] != nil:
 				return f.openAIChatMessages(v), FormatterOpenAIChat
 			}
 		}
 	case map[string]any:
+		if inner, ok := superJSON(v); ok {
+			return f.side(decode(inner), role)
+		}
 		switch {
 		case isLangChainRun(v):
 			return f.langChainRun(v), FormatterLangChain
@@ -214,6 +222,9 @@ func (f *formatting) side(value any, role Role) (OutputMessages, string) {
 		if messages, ok := v["messages"].([]any); ok {
 			return f.openAIChatMessages(messages), FormatterOpenAIChat
 		}
+		if (v["role"] != nil && hasKeys(v, "content")) || v["tool_calls"] != nil {
+			return OutputMessages{f.openAIChatSingle(v)}, FormatterOpenAIChat
+		}
 	case string:
 		if v == "" {
 			return OutputMessages{}, FormatterText
@@ -309,6 +320,23 @@ func foldAssistant(msgs OutputMessages) OutputMessages {
 	return out
 }
 
+// superJSON unwraps {json, meta?}, the envelope superjson, the transformer tRPC apps serialise with,
+// puts around a value; a Langfuse generation records it around the request when the app passes the
+// serialised form:
+// https://github.com/flightcontrolhq/superjson/blob/aaa65e36e8e31da740265a56b27c965ca1c7b754/README.md
+func superJSON(m map[string]any) (any, bool) {
+	inner, ok := m["json"]
+	if !ok {
+		return nil, false
+	}
+	for key := range m {
+		if key != "json" && key != "meta" {
+			return nil, false
+		}
+	}
+	return inner, true
+}
+
 // decode parses a JSON-encoded attribute value, twice when an SDK encoded the string again.
 // Anything that is not JSON is returned as is.
 func decode(value any) any {

pkg/types/aiobservabilitytypes/message_format_openai_chat.go

@@ -4,8 +4,9 @@ import (
 	"strings"
 )
 
-// Format: OpenAI Chat Completions, a request {messages}, a response {choices}, or a bare message list.
-// Written by: bifrost, openrouter requests, the raw OpenAI calls OpenInference records.
+// Format: OpenAI Chat Completions, a request {messages}, a response {choices}, a bare message list, or a
+// single message.
+// Written by: bifrost, openrouter requests, the raw OpenAI calls OpenInference records, the Langfuse SDK.
 // Handled: string and block content, tool_calls nested or flattened, tool messages, refusal,
 // reasoning_content, finish reasons.
 // Not yet: streaming delta chunks, legacy completions choices[].text, audio output.
@@ -88,6 +89,19 @@ func (f *formatting) openAIChatMessage(m map[string]any) OutputMessage {
 	return msg
 }
 
+// openAIChatSingle converts a message recorded outside a list: the first choice's message the Langfuse
+// OpenAI wrapper writes as the output, or the additional_kwargs of a LangChain AIMessage, {tool_calls}
+// without a role, which the Langfuse LangChain handler writes for a tool-calling turn:
+// https://github.com/langfuse/langfuse-python/blob/18baf80450dd5f9c1f85916eb2d1b3c72d2052e1/langfuse/openai.py#L446
+// https://github.com/langfuse/langfuse-python/blob/18baf80450dd5f9c1f85916eb2d1b3c72d2052e1/langfuse/langchain/CallbackHandler.py#L1558
+func (f *formatting) openAIChatSingle(m map[string]any) OutputMessage {
+	msg := f.openAIChatMessage(m)
+	if msg.Role.IsZero() {
+		msg.Role = RoleAssistant
+	}
+	return msg
+}
+
 // toolCallPart reads {id, function: {name, arguments}}; a gateway may flatten it to {id, name, arguments|args}.
 func toolCallPart(c map[string]any) Part {
 	call := ToolCallRequestPart{Type: PartTypeToolCall, ID: optionalString(c["id"]), Name: stringOf(c["name"])}

pkg/types/aiobservabilitytypes/message_format_openai_chat_test.go

@@ -205,6 +205,30 @@ func TestFormat_OpenAIChat(t *testing.T) {
 			]`,
 			wantOutput: `[]`,
 		},
+		{
+			// Langfuse SDK generation: the history as a list, the first choice's message as the output
+			name:       "SingleMessageOutput_AssistantMessage",
+			input:      `[{"role": "system", "content": "You are a support agent on a phone call."}, {"role": "user", "content": "Hello?"}]`,
+			output:     `{"role": "assistant", "content": "Hi, this is a courtesy call about your order. Do you have a minute?"}`,
+			wantInput:  `[{"role": "system", "parts": [{"type": "text", "content": "You are a support agent on a phone call."}]}, {"role": "user", "parts": [{"type": "text", "content": "Hello?"}]}]`,
+			wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "Hi, this is a courtesy call about your order. Do you have a minute?"}]}]`,
+		},
+		{
+			// a structured output that happens to carry a role key is not a message
+			name:       "ObjectWithRoleKeyButNoContent_KeptAsGeneric",
+			output:     `{"role": "admin", "permissions": ["read", "write"]}`,
+			warnings:   []string{"message format not recognised, kept as generic"},
+			wantInput:  `[]`,
+			wantOutput: `[{"role": "", "parts": [{"type": "generic", "content": "{\"permissions\":[\"read\",\"write\"],\"role\":\"admin\"}"}]}]`,
+		},
+		{
+			// Langfuse LangChain handler: the additional_kwargs of the AIMessage, without a role, as the output
+			// of a tool-calling turn
+			name:       "RolelessToolCallsOutput_AssistantToolCallPart",
+			output:     `{"tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": "{\"city\": \"Oslo\"}"}}], "refusal": null}`,
+			wantInput:  `[]`,
+			wantOutput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "id": "call_1", "arguments": {"city": "Oslo"}}]}]`,
+		},
 		{
 			name:      "ResponseWithScalarChoice_KeptAsGeneric",
 			output:    `{"choices":["oops",{"finish_reason":"stop","index":1,"message":{"role":"assistant","content":"hi"}}]}`,

pkg/types/aiobservabilitytypes/message_format_openai_responses.go

@@ -2,13 +2,15 @@ package aiobservabilitytypes
 
 import "strings"
 
-// Format: OpenAI Responses, a request {input, instructions} or a response {output, status} of items:
+// Format: OpenAI Responses, a request {input, instructions}, a response {output, status} or a bare list
+// of items:
 // https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L47597
 // https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L69691
 // https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L53834
 // https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L57596
-// Written by: gateways recording raw Responses bodies, and the raw calls OpenInference records for the
-// Responses API and the OpenAI Agents SDK when its flattened messages are absent.
+// Written by: gateways recording raw Responses bodies, the raw calls OpenInference records for the
+// Responses API and the OpenAI Agents SDK when its flattened messages are absent, and LiteLLM's
+// langfuse_otel callback, the function_call items of a tool-calling turn as a bare list.
 // Handled: message items with input_text, output_text, refusal, input_image and input_file content;
 // function_call and custom_tool_call with their outputs; reasoning summaries; compaction; every other
 // *_call item as a server tool call and *_output item as its response; the status and incomplete_details
@@ -32,6 +34,19 @@ func isResponsesRequest(m map[string]any) bool {
 	return isItem
 }
 
+// isResponsesItems looks for an item only the Responses API has, a tool call or its output, a reasoning
+// or compaction item, all without a role; a list of messages has none.
+func isResponsesItems(list []any) bool {
+	for _, item := range list {
+		m, _ := item.(map[string]any)
+		typ := stringOf(m["type"])
+		if m["role"] == nil && (typ == "reasoning" || typ == "compaction" || typ == "mcp_approval_response" || strings.HasSuffix(typ, "_call") || strings.HasSuffix(typ, "_output")) {
+			return true
+		}
+	}
+	return false
+}
+
 func isResponsesResponse(m map[string]any) bool {
 	if _, ok := m["output"].([]any); !ok {
 		return false

pkg/types/aiobservabilitytypes/message_format_openai_responses_test.go

@@ -73,6 +73,23 @@ func TestFormat_OpenAIResponses(t *testing.T) {
 			wantInput:  `[{"role": "system", "parts": [{"type": "text", "content": "Be brief."}]}, {"role": "user", "parts": [{"type": "text", "content": "Say hi"}]}]`,
 			wantOutput: `[]`,
 		},
+		{
+			// LiteLLM langfuse_otel: the function_call items of a tool-calling turn as a bare list
+			name:       "BareFunctionCallItems_OneAssistantMessage",
+			output:     `[{"id": "chatcmpl-1", "name": "set_stage", "call_id": "call_7", "type": "function_call", "arguments": {"fields": {"stage": "consent"}}}]`,
+			wantInput:  `[]`,
+			wantOutput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "set_stage", "id": "call_7", "arguments": {"fields": {"stage": "consent"}}}]}]`,
+		},
+		{
+			name:  "BareItemsWithMessage_ToolRound",
+			input: `[{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "Weather in Oslo?"}]}, {"type": "function_call", "call_id": "call_9", "name": "get_weather", "arguments": "{\"city\":\"Oslo\"}"}, {"type": "function_call_output", "call_id": "call_9", "output": "4C, cloudy"}]`,
+			wantInput: `[
+			  {"role": "user", "parts": [{"type": "text", "content": "Weather in Oslo?"}]},
+			  {"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "id": "call_9", "arguments": {"city": "Oslo"}}]},
+			  {"role": "tool", "parts": [{"type": "tool_call_response", "id": "call_9", "response": "4C, cloudy"}]}
+			]`,
+			wantOutput: `[]`,
+		},
 		{
 			// openinference capture, CreateEmbeddings span: an embeddings request also has input and model
 			name:       "EmbeddingsRequest_NotResponses_KeptAsGeneric",

pkg/types/aiobservabilitytypes/message_format_openinference.go

@@ -4,8 +4,9 @@ import "github.com/SigNoz/signoz/pkg/valuer"
 
 // Format: the flattened llm.input_messages.N.message.* and llm.output_messages.N.message.* keys, the
 // provider-neutral copy of the conversation the OpenInference instrumentations write on LLM spans. A side
-// without them falls back to the raw provider payload the mapper moved onto gen_ai.*.messages; pipecat
-// streams the reply into output.value alone.
+// without them, or whose buckets hold no message, falls back to the raw provider payload the mapper moved
+// onto gen_ai.*.messages; pipecat streams the reply into output.value alone, LiteLLM's OpenInference mode
+// leaves the output bucket empty on a tool-calling turn.
 // https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/spec/semantic_conventions.md#L77
 // https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/spec/tool_calling.md#L37
 // https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/spec/multimodal_attributes.md#L35
@@ -28,15 +29,14 @@ func (f *formatting) openInference() bool {
 		return false
 	}
 	inputLabel, outputLabel := FormatterOpenInference, FormatterOpenInference
-	if len(input) > 0 {
-		f.out.Input = inputMessages(openInferenceMessages(input))
+	if msgs := openInferenceMessages(input); len(msgs) > 0 {
+		f.out.Input = inputMessages(msgs)
 	} else if raw := attribute(f.attributes, GenAIInputMessages); raw != nil {
-		var msgs OutputMessages
 		msgs, inputLabel = f.side(decode(raw), RoleUser)
 		f.out.Input = inputMessages(msgs)
 	}
-	if len(output) > 0 {
-		f.out.Output = openInferenceMessages(output)
+	if msgs := openInferenceMessages(output); len(msgs) > 0 {
+		f.out.Output = msgs
 		if fr := finishReason(stringOf(attribute(f.attributes, openInferenceFinishReason))); fr != nil {
 			for i := range f.out.Output {
 				f.out.Output[i].FinishReason = fr

pkg/types/aiobservabilitytypes/message_format_openinference_test.go

@@ -160,16 +160,30 @@ func TestFormat_OpenInference(t *testing.T) {
 			// which the mapper moves onto gen_ai.output.messages
 			// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/python/instrumentation/openinference-instrumentation-pipecat/src/openinference/instrumentation/pipecat/_attributes.py#L430
 			// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/python/instrumentation/openinference-instrumentation-pipecat/src/openinference/instrumentation/pipecat/_observer.py#L628
-			name: "OutputKeysAbsent_RawOutputFormatted_Warns",
+			name: "OutputKeysAbsent_RawOutputFormatted",
 			attributes: map[string]any{
 				"llm.input_messages.0.message.role":    "user",
 				"llm.input_messages.0.message.content": "hi",
 				GenAIOutputMessages:                    "Hello there.",
 			},
-			warnings:   []string{"bare assistant text, role assumed", "output formatted as text, input as openinference"},
+			warnings:   []string{"bare assistant text, role assumed"},
 			wantInput:  `[{"role": "user", "parts": [{"type": "text", "content": "hi"}]}]`,
 			wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "Hello there."}]}]`,
 		},
+		{
+			// LiteLLM's OpenInference mode leaves the output bucket empty on a tool-calling turn; the calls are
+			// the Langfuse output the mapper moved onto gen_ai.output.messages
+			name: "EmptyOutputBucket_RawOutputFormatted",
+			attributes: map[string]any{
+				"llm.input_messages.0.message.role":     "user",
+				"llm.input_messages.0.message.content":  "Yes, go ahead.",
+				"llm.output_messages.0.message.role":    "assistant",
+				"llm.output_messages.0.message.content": "",
+				GenAIOutputMessages:                     `[{"id": "chatcmpl-2", "name": "set_stage", "call_id": "call_8", "type": "function_call", "arguments": {"fields": {"stage": "email"}}}]`,
+			},
+			wantInput:  `[{"role": "user", "parts": [{"type": "text", "content": "Yes, go ahead."}]}]`,
+			wantOutput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "set_stage", "id": "call_8", "arguments": {"fields": {"stage": "email"}}}]}]`,
+		},
 		{
 			// message.role left unset by some instrumentors: tool_call_id, tool_calls or neither decide it,
 			// as the collector-contrib normalizer does

pkg/types/aiobservabilitytypes/message_format_test.go

@@ -122,6 +122,31 @@ So, the ball costs 5 cents.`,
 			wantInput:  `[]`,
 			wantOutput: `[]`,
 		},
+		{
+			// Langfuse SDK generation: the app passes the request through superjson, the output is the parsed
+			// structured output
+			name:       "SuperJSONEnvelope_Unwrapped_StructuredOutputGeneric",
+			input:      `{"json": {"model": "gpt-4o-mini", "messages": [{"role": "system", "content": "Extract the main article of the page as JSON."}, {"role": "user", "content": "# Council approves new bike lanes"}], "temperature": 0}}`,
+			output:     `{"hasArticle": true, "article": {"title": "Council approves new bike lanes"}}`,
+			warnings:   []string{"message format not recognised, kept as generic"},
+			wantInput:  `[{"role": "system", "parts": [{"type": "text", "content": "Extract the main article of the page as JSON."}]}, {"role": "user", "parts": [{"type": "text", "content": "# Council approves new bike lanes"}]}]`,
+			wantOutput: `[{"role": "", "parts": [{"type": "generic", "content": "{\"article\":{\"title\":\"Council approves new bike lanes\"},\"hasArticle\":true}"}]}]`,
+		},
+		{
+			name:       "SuperJSONEnvelopeWithMeta_Unwrapped",
+			input:      `{"json": {"messages": [{"role": "user", "content": "What time is it in Oslo?"}]}, "meta": {"values": {"messages.0.at": ["Date"]}}}`,
+			output:     `{"json": {"choices": [{"finish_reason": "stop", "message": {"role": "assistant", "content": "It is 14:05 in Oslo."}}]}, "meta": {"values": {}}}`,
+			wantInput:  `[{"role": "user", "parts": [{"type": "text", "content": "What time is it in Oslo?"}]}]`,
+			wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "It is 14:05 in Oslo."}], "finish_reason": "stop"}]`,
+		},
+		{
+			// a json key beside others is a payload of its own, not the envelope
+			name:       "JSONKeyBesideOthers_NotUnwrapped_Generic",
+			input:      `{"json": {"messages": []}, "schema": "v1"}`,
+			warnings:   []string{"message format not recognised, kept as generic"},
+			wantInput:  `[{"role": "", "parts": [{"type": "generic", "content": "{\"json\":{\"messages\":[]},\"schema\":\"v1\"}"}]}]`,
+			wantOutput: `[]`,
+		},
 		{
 			name:       "NoMessageAttributes_EmptyLists",
 			attributes: map[string]any{GenAIRequestModel: "gpt-4o", GenAIToolName: "get_weather"},

pkg/types/querybuildertypes/querybuildertypesv5/req.go

@@ -664,12 +664,6 @@ func (r *QueryRangeRequest) UnmarshalJSON(data []byte) error {
 	// Copy the decoded values back to the original struct
 	*r = QueryRangeRequest(temp)
 
-	// Coerce the window to epoch milliseconds up front so every downstream
-	// consumer (TimeRange, narrowWindowByTraceID, step interval, etc.) can
-	// safely assume ms regardless of the resolution the caller sent.
-	r.Start = ToMilliSecs(r.Start)
-	r.End = ToMilliSecs(r.End)
-
 	return nil
 }
 

pkg/types/querybuildertypes/querybuildertypesv5/time.go

@@ -1,25 +0,0 @@
-package querybuildertypesv5
-
-import "math"
-
-// ToMilliSecs takes an epoch whose resolution is inferred from its magnitude
-// (s/ms/µs/ns) and returns it in milliseconds. A millisecond epoch for the
-// current era has 13 digits (e.g. ~1.7e12 in 2026), so the value is scaled so
-// its digit-width matches: smaller values (seconds) are scaled up, larger ones
-// (micro/nanoseconds) are scaled down. Zero is returned unchanged.
-func ToMilliSecs(epoch uint64) uint64 {
-	if epoch == 0 {
-		return 0
-	}
-	temp := epoch
-	count := 0
-	for epoch != 0 {
-		epoch /= 10
-		count++
-	}
-	const msDigits = 13
-	if count < msDigits {
-		return temp * uint64(math.Pow(10, float64(msDigits-count)))
-	}
-	return temp / uint64(math.Pow(10, float64(count-msDigits)))
-}

pkg/types/querybuildertypes/querybuildertypesv5/time_test.go

@@ -1,52 +0,0 @@
-package querybuildertypesv5
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestToMilliSecs(t *testing.T) {
-	testCases := []struct {
-		name     string
-		epoch    uint64
-		expected uint64
-	}{
-		{
-			name:     "SecondsEpoch_TenDigits_ScaledUp",
-			epoch:    1672531200,    // seconds, 2023-01-01 00:00:00 UTC
-			expected: 1672531200000, // * 10^3
-		},
-		{
-			name:     "MilliSecondsEpoch_ThirteenDigits_Unchanged",
-			epoch:    1672531200000,
-			expected: 1672531200000, // unchanged
-		},
-		{
-			name:     "MicroSecondsEpoch_SixteenDigits_ScaledDown",
-			epoch:    1672531200000000, // microseconds
-			expected: 1672531200000,    // / 10^3
-		},
-		{
-			name:     "NanoSecondsEpoch_NineteenDigits_ScaledDown",
-			epoch:    1672531200000000000, // nanoseconds
-			expected: 1672531200000,       // / 10^6
-		},
-		{
-			name:     "Zero_Unchanged",
-			epoch:    0,
-			expected: 0,
-		},
-		{
-			name:     "SecondsEpoch_RecentTimestamp_ScaledUp",
-			epoch:    1716638400, // 2024-05-25 12:00:00 UTC
-			expected: 1716638400000,
-		},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			assert.Equal(t, testCase.expected, ToMilliSecs(testCase.epoch))
-		})
-	}
-}

pkg/types/querybuildertypes/querybuildertypesv5/validation.go

@@ -5,16 +5,12 @@ import (
 	"math"
 	"slices"
 	"strings"
-	"time"
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 )
 
-// maxEndTimeAhead tolerates client clock skew; anything further ahead is a malformed timestamp.
-const maxEndTimeAhead = 24 * time.Hour
-
 // getQueryIdentifier returns a friendly identifier for a query based on its type and name/content.
 func getQueryIdentifier(envelope QueryEnvelope, index int) string {
 	name := envelope.GetQueryName()
@@ -578,8 +574,12 @@ func (q *QueryBuilderQuery[T]) validateOrderByForAggregation() error {
 
 // Validate validates the entire query range request.
 func (r *QueryRangeRequest) Validate(opts ...ValidationOption) error {
-	if err := r.validateTimeRange(); err != nil {
-		return err
+	// Validate time range
+	if r.RequestType != RequestTypeRawStream && r.Start >= r.End {
+		return errors.NewInvalidInputf(
+			errors.CodeInvalidInput,
+			"start time must be before end time",
+		)
 	}
 
 	// Validate request type
@@ -631,8 +631,8 @@ func (r *QueryRangeRequest) Validate(opts ...ValidationOption) error {
 // this so per-query errors can be attributed individually via QueryEnvelope.Validate
 // instead of failing fast like Validate does.
 func (r *QueryRangeRequest) ValidateRequestScope() ([]ValidationOption, error) {
-	if err := r.validateTimeRange(); err != nil {
-		return nil, err
+	if r.RequestType != RequestTypeRawStream && r.Start >= r.End {
+		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "start time must be before end time")
 	}
 
 	var opts []ValidationOption
@@ -699,22 +699,6 @@ func (e QueryEnvelope) Validate(opts ...ValidationOption) error {
 	return validateQueryEnvelope(e, opts...)
 }
 
-func (r *QueryRangeRequest) validateTimeRange() error {
-	if r.RequestType == RequestTypeRawStream {
-		return nil
-	}
-	if r.Start == 0 {
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "start time is required")
-	}
-	if r.Start >= r.End {
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "start time must be before end time")
-	}
-	if r.End > uint64(time.Now().Add(maxEndTimeAhead).UnixMilli()) {
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "end time %s is too far in the future", time.UnixMilli(int64(r.End)).UTC().Format(time.RFC3339))
-	}
-	return nil
-}
-
 // validateAllQueriesNotDisabled validates that at least one query in the composite query is enabled.
 func (r *QueryRangeRequest) validateAllQueriesNotDisabled() error {
 	for _, envelope := range r.CompositeQuery.Queries {

tests/integration/tests/queriercommon/09_time_range.py

@@ -1,165 +0,0 @@
-from collections.abc import Callable
-from datetime import UTC, datetime, timedelta
-from http import HTTPStatus
-
-import pytest
-
-from fixtures import types
-from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
-from fixtures.logs import Logs
-from fixtures.metrics import Metrics
-from fixtures.querier import (
-    Aggregation,
-    BuilderQuery,
-    MetricAggregation,
-    get_all_warnings,
-    get_error_message,
-    get_series_values,
-    make_query_request,
-)
-from fixtures.traces import TraceIdGenerator, Traces
-
-TWO_YEARS = timedelta(days=730)
-
-
-@pytest.mark.parametrize(
-    "start_ms,end_ms,message",
-    [
-        pytest.param(0, 1_700_000_000_000, "start time is required", id="zero_start"),
-        pytest.param(1_700_000_000_000, 1_700_000_000_000, "start time must be before end time", id="start_equals_end"),
-        # 2592000000 is 30 days in ms; read as an epoch in seconds it is year 2052.
-        pytest.param(1_700_000_000_000, 2_592_000_000, "end time 2052-02-20T00:00:00Z is too far in the future", id="end_in_2052"),
-    ],
-)
-def test_time_range_rejected(
-    start_ms: int,
-    end_ms: int,
-    message: str,
-    signoz: types.SigNoz,
-    create_user_admin: None,  # pylint: disable=unused-argument
-    get_token: Callable[[str, str], str],
-) -> None:
-    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-
-    response = make_query_request(
-        signoz,
-        token,
-        start_ms=start_ms,
-        end_ms=end_ms,
-        queries=[BuilderQuery(signal="logs", aggregations=[Aggregation(expression="count()")], step_interval=60).to_dict()],
-    )
-
-    assert response.status_code == HTTPStatus.BAD_REQUEST
-    assert get_error_message(response.json()) == message
-
-
-def test_two_year_logs_timeseries_step_clamped(
-    signoz: types.SigNoz,
-    create_user_admin: None,  # pylint: disable=unused-argument
-    get_token: Callable[[str, str], str],
-    insert_logs: Callable[[list[Logs]], None],
-) -> None:
-    # A 1s step over 2 years is clamped to 42045s (at most ~1500 points); end is aligned to it so the bucket count is stable.
-    step_seconds = 42045
-    end = datetime.fromtimestamp(int(datetime.now(tz=UTC).timestamp()) // step_seconds * step_seconds, tz=UTC)
-    insert_logs([Logs(timestamp=end - timedelta(minutes=1), resources={"service.name": "time-range-logs"}, body="two year span")])
-
-    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-    response = make_query_request(
-        signoz,
-        token,
-        start_ms=int((end - TWO_YEARS).timestamp() * 1000),
-        end_ms=int(end.timestamp() * 1000),
-        queries=[
-            BuilderQuery(
-                signal="logs",
-                filter_expression="service.name = 'time-range-logs'",
-                aggregations=[Aggregation(expression="count()")],
-                step_interval=1,
-            ).to_dict()
-        ],
-        format_options={"formatTableResultForUI": False, "fillGaps": True},
-    )
-
-    assert response.status_code == HTTPStatus.OK
-    assert response.json()["data"]["meta"]["stepIntervals"] == {"A": step_seconds}
-    assert get_all_warnings(response.json()) == [
-        {"message": f"Query A is requesting aggregation interval 1 seconds, which is smaller than the minimum allowed interval of {step_seconds} seconds for selected time range. Using the minimum instead"},
-    ]
-    values = get_series_values(response.json(), "A")
-    assert len(values) == 1502
-    assert sum(v["value"] for v in values) == 1
-
-
-def test_two_year_traces_timeseries_step_clamped(
-    signoz: types.SigNoz,
-    create_user_admin: None,  # pylint: disable=unused-argument
-    get_token: Callable[[str, str], str],
-    insert_traces: Callable[[list[Traces]], None],
-) -> None:
-    # Same clamp as logs: 42045s; end is aligned to it so the bucket count is stable.
-    step_seconds = 42045
-    end = datetime.fromtimestamp(int(datetime.now(tz=UTC).timestamp()) // step_seconds * step_seconds, tz=UTC)
-    insert_traces([Traces(timestamp=end - timedelta(minutes=1), trace_id=TraceIdGenerator.trace_id(), span_id=TraceIdGenerator.span_id(), resources={"service.name": "time-range-traces"})])
-
-    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-    response = make_query_request(
-        signoz,
-        token,
-        start_ms=int((end - TWO_YEARS).timestamp() * 1000),
-        end_ms=int(end.timestamp() * 1000),
-        queries=[
-            BuilderQuery(
-                signal="traces",
-                filter_expression="service.name = 'time-range-traces'",
-                aggregations=[Aggregation(expression="count()")],
-                step_interval=1,
-            ).to_dict()
-        ],
-        format_options={"formatTableResultForUI": False, "fillGaps": True},
-    )
-
-    assert response.status_code == HTTPStatus.OK
-    assert response.json()["data"]["meta"]["stepIntervals"] == {"A": step_seconds}
-    assert get_all_warnings(response.json()) == [
-        {"message": f"Query A is requesting aggregation interval 1 seconds, which is smaller than the minimum allowed interval of {step_seconds} seconds for selected time range. Using the minimum instead"},
-    ]
-    values = get_series_values(response.json(), "A")
-    assert len(values) == 1502
-    assert sum(v["value"] for v in values) == 1
-
-
-def test_two_year_metrics_timeseries_step_clamped(
-    signoz: types.SigNoz,
-    create_user_admin: None,  # pylint: disable=unused-argument
-    get_token: Callable[[str, str], str],
-    insert_metrics: Callable[[list[Metrics]], None],
-) -> None:
-    # A 1s step over 2 years is clamped to 210600s (~300 points); end is aligned to it so the bucket count is stable.
-    step_seconds = 210600
-    metric_name = "time_range_two_year_gauge"
-    end = datetime.fromtimestamp(int(datetime.now(tz=UTC).timestamp()) // step_seconds * step_seconds, tz=UTC)
-    insert_metrics([Metrics(metric_name=metric_name, labels={"service": "time-range-metrics"}, timestamp=end - timedelta(minutes=1), value=1.0, temporality="Unspecified", type_="Gauge", is_monotonic=False)])
-
-    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-    response = make_query_request(
-        signoz,
-        token,
-        start_ms=int((end - TWO_YEARS).timestamp() * 1000),
-        end_ms=int(end.timestamp() * 1000),
-        queries=[
-            BuilderQuery(
-                signal="metrics",
-                aggregations=[MetricAggregation(metric_name=metric_name, time_aggregation="latest", space_aggregation="sum", temporality="unspecified")],
-                step_interval=1,
-            ).to_dict()
-        ],
-        format_options={"formatTableResultForUI": False, "fillGaps": True},
-    )
-
-    assert response.status_code == HTTPStatus.OK
-    assert response.json()["data"]["meta"]["stepIntervals"] == {"A": step_seconds}
-    assert get_all_warnings(response.json()) == [
-        {"message": f"Query A is requesting aggregation interval 1 seconds, which is smaller than the minimum allowed interval of {step_seconds} seconds for selected time range. Using the minimum instead"},
-    ]
-    assert len(get_series_values(response.json(), "A")) == 301