Merge branch 'main' into feat/ai-attr-mapping-updates

#### Description

- Opamp places user log pipelines around the first
`signozlogsnormalizer` in the logs pipeline, per its `opamp.position`:
the normalizer runs after them by default (`after`), before them with
`before`. Other processors keep their places.
- Without the normalizer, processor ordering is unchanged.
- The `use_json_body`-prepended normalize operator is dropped when the
normalizer already runs ahead of user pipelines (`before`); otherwise
both are kept.
- Removes the `json_body_dual_ingestion` flag: the normalize operator no
longer stashes the original body, and isn't appended after user
pipelines when only that flag is on.

| `use_json_body` | `opamp.position` | Logs pipeline | Normalize
operator |
|---|---|---|---|
| off | no normalizer | memory_limiter → user → custom → batch
(unchanged) | — |
| off | `after` (default) | … → user → normalizer → … | — |
| off | `before` | … → normalizer → user → … | — |
| on | no normalizer | memory_limiter → normalize op → user → custom →
batch | prepended |
| on | `after` (default) | … → normalize op → user → normalizer → … |
prepended (both normalize) |
| on | `before` | … → normalizer → user → … | dropped |

#### Additional Information

- Depends on SigNoz/signoz-otel-collector#953 for the `opamp.position`
config.
- The pipelines preview still prepends the operator under
`use_json_body`, since it can't see per-collector `opamp.position`.
- Configs still setting `json_body_dual_ingestion` keep working; the
unknown flag is skipped with a warning.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

.claude/rules/pytest.md

@@ -17,3 +17,4 @@ For the Python integration suite under `tests/`. Setup, running, and suite layou
 - **snake_case parametrize ids.** `ids=["fill_gaps", "fill_zero"]`, not camelCase.
 - **Name suite files with the two-digit prefix (`NN_*.py`).** `pyproject.toml` restricts collection to `[0-9][0-9]_*.py` (plus the bootstrap `setup.py` / `run.py`) — a file that doesn't match is silently never collected.
 - **Always run pytest from `tests/`.** `--import-mode=importlib` is what allows same-basename files across suites (`querier/01_logs.py` vs `rawexportdata/01_logs.py`), but it disables pytest's implicit `sys.path` injection — `import fixtures` only resolves via `pythonpath = ["."]` from that rootdir.
+- **After changing any `.py` file, run `make py-fmt` and `make py-lint`** from the repo root and fix what they report.

pkg/flagger/registry.go

@@ -9,7 +9,6 @@ var (
 	FeaturePutMetersInZeus        = featuretypes.MustNewName("put_meters_in_zeus")
 	FeatureUseMeterReporter       = featuretypes.MustNewName("use_meter_reporter")
 	FeatureUseJSONBody            = featuretypes.MustNewName("use_json_body")
-	FeatureJSONBodyDualIngestion  = featuretypes.MustNewName("json_body_dual_ingestion")
 	FeatureEnableMetricsReduction = featuretypes.MustNewName("enable_metrics_reduction")
 	FeatureResolveSemconvFamilies = featuretypes.MustNewName("resolve_semconv_families")
 	FeatureUseTraceAttributesJSON = featuretypes.MustNewName("use_trace_attributes_json")
@@ -65,14 +64,6 @@ func MustNewRegistry() featuretypes.Registry {
 			DefaultVariant: featuretypes.MustNewName("disabled"),
 			Variants:       featuretypes.NewBooleanVariants(),
 		},
-		&featuretypes.Feature{
-			Name:           FeatureJSONBodyDualIngestion,
-			Kind:           featuretypes.KindBoolean,
-			Stage:          featuretypes.StageExperimental,
-			Description:    "Controls whether the collector's normalize operator keeps the original log body so it is ingested into both the legacy body and the JSON body columns",
-			DefaultVariant: featuretypes.MustNewName("disabled"),
-			Variants:       featuretypes.NewBooleanVariants(),
-		},
 		&featuretypes.Feature{
 			Name:           FeatureEnableMetricsReduction,
 			Kind:           featuretypes.KindBoolean,

pkg/modules/spanmapper/implspanmapper/fs/definitions/README.md

@@ -0,0 +1,62 @@
+# Shipped span mapper definitions
+
+Where each source key comes from. A group runs on spans with an attribute key containing one of its condition substrings; sources are tried by priority and the first present wins; an existing target is never overwritten. Bump `version` on every change.
+
+## gen_ai.llm
+
+Condition: `model`, `llm.input_messages` (OpenInference sets `llm.model_name` from the response, so a failed call has no model key).
+
+| Source | Emitted by |
+|---|---|
+| `llm.model_name`, `llm.provider`, `llm.system`, `llm.token_count.prompt`, `llm.token_count.completion`, `llm.token_count.prompt_details.cache_read`, `llm.token_count.prompt_details.cache_write`, `llm.token_count.completion_details.reasoning`, `llm.finish_reason`, `input.value`, `output.value`, `embedding.model_name`, `reranker.model_name`, `openinference.span.kind` | OpenInference |
+| `llm.request.model`, `llm.response.model`, `llm.vendor`, `llm.usage.prompt_tokens`, `llm.usage.completion_tokens`, `llm.response.finish_reason`, `llm.response.stop_reason` | OpenLLMetry before the `gen_ai` rename |
+| `gen_ai.system`, `gen_ai.usage.prompt_tokens`, `gen_ai.usage.completion_tokens`, `gen_ai.usage.cache_read_input_tokens`, `gen_ai.usage.cache_creation_input_tokens`, `gen_ai.usage.reasoning_tokens`, `gen_ai.prompt`, `gen_ai.completion`, `llm.request.type`, `traceloop.span.kind` | OpenLLMetry, LiteLLM, CrewAI, Mistral, LangSmith, OpenRouter gateway |
+| `gen_ai.usage.cache_write.input_tokens` | OTel semconv draft spelling, Codex |
+| `gen_ai.response.finish_reason`, `gen_ai.usage.prompt_token_details.cached_read_tokens`, `gen_ai.usage.completion_token_details.reasoning_tokens` | Bifrost gateway |
+| `gen_ai.usage.input_tokens.cached` | OpenRouter gateway |
+| `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 |
+| `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)) |
+
+## gen_ai.tool
+
+Condition: `tool.name` (OpenInference), `ai.toolCall.` (Vercel, Mastra), `tool_call_args`, `tool_response` (Google ADK), `tool_use_id` (Claude Code, opentelemetry-hooks). Not `tool_name`: Codex records it on the model call span.
+
+| Source | Emitted by |
+|---|---|
+| `tool.name`, `tool.id`, `tool.description`, `input.value`, `output.value`, `openinference.span.kind` | OpenInference |
+| `ai.toolCall.name`, `ai.toolCall.id`, `ai.toolCall.args`, `ai.toolCall.result` | Vercel AI SDK, Mastra |
+| `traceloop.entity.input`, `traceloop.entity.output`, `traceloop.span.kind`, `llm.request.type` | OpenLLMetry |
+| `gcp.vertex.agent.tool_call_args`, `gcp.vertex.agent.tool_response` | Google ADK |
+| `tool_name` | Claude Code `claude_code.tool` spans |
+| `gen_ai.client.tool_name`, `gen_ai.client.tool_use_id` | opentelemetry-hooks runner (Cursor, Copilot, Claude Code hooks) |
+
+## gen_ai.agent
+
+Condition: `agent`.
+
+| Source | Emitted by |
+|---|---|
+| `agent.name`, `agent.id`, `agent.description`, `openinference.span.kind` | OpenInference; `agent.id` and `agent.description` seen in no capture |
+| `agent_name`, `final_result` | Pydantic AI |
+| `crewai.agent.role`, `crewai.agent.id` | CrewAI |
+| `traceloop.span.kind`, `llm.request.type` | OpenLLMetry |
+| `agent_id` | Claude Code |
+
+## Coverage of documented integrations
+
+Every integration under [signoz.io/docs/llm-observability](https://signoz.io/docs/llm-observability/) was checked by looking at the instrumentation its page installs and the keys that instrumentation writes.
+
+- Most pages install OpenInference or OpenLLMetry, or an SDK that writes OTel GenAI semconv itself (OpenAI, OpenAI Agents, Pydantic AI, AutoGen, Microsoft Agent Framework, Semantic Kernel, Copilot Chat, LiteLLM, OpenRouter). Their keys are the rows above.
+- Three write their own keys and have their own rows: Claude Code and Claude Agent SDK, Cursor through opentelemetry-hooks, vLLM (response model only).
+- Codex cannot be priced by renaming: `model` sits on the sampling function spans, `gen_ai.usage.*` on the child `handle_responses` span, `conversation.id` on the `codex.sampling` phase span, never together on one span.
+
+To check a new integration, capture a trace and look for a model, usage, tool or messages key that no row above lists.
+
+## Left unmapped on purpose
+
+LiteLLM `llm.openai.messages` and `llm.openai.choices` (debug echo of the parent span's messages, Python repr). Total tokens and vendor cost keys (no target; the pricing processor owns cost). `ai.response.msToFirstChunk` and `gen_ai.latency.time_to_first_token` (milliseconds and seconds under one target).

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

@@ -1,5 +1,5 @@
 {
-  "version": 2,
+  "version": 3,
   "definition": {
     "name": "gen_ai.agent",
     "condition": {
@@ -17,6 +17,12 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "crewai.agent.role",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 30
+            },
             {
               "key": "agent.name",
               "context": "attribute",
@@ -37,6 +43,18 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "agent_id",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 30
+            },
+            {
+              "key": "crewai.agent.id",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 20
+            },
             {
               "key": "agent.id",
               "context": "attribute",
@@ -73,6 +91,32 @@
             }
           ]
         }
+      },
+      {
+        "name": "gen_ai.operation.name",
+        "fieldContext": "attribute",
+        "config": {
+          "sources": [
+            {
+              "key": "openinference.span.kind",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 30
+            },
+            {
+              "key": "llm.request.type",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 20
+            },
+            {
+              "key": "traceloop.span.kind",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 10
+            }
+          ]
+        }
       }
     ]
   }

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

@@ -1,11 +1,14 @@
 {
-  "version": 3,
+  "version": 4,
   "definition": {
     "name": "gen_ai.llm",
     "condition": {
       "attributes": [
         {
           "value": "model"
+        },
+        {
+          "value": "llm.input_messages"
         }
       ],
       "resource": []
@@ -17,6 +20,12 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "reranker.model_name",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 70
+            },
             {
               "key": "llm.model_name",
               "context": "attribute",
@@ -52,6 +61,12 @@
               "context": "attribute",
               "operation": "copy",
               "priority": 10
+            },
+            {
+              "key": "gen_ai.response.model",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 5
             }
           ]
         }
@@ -114,11 +129,43 @@
           ]
         }
       },
+      {
+        "name": "gen_ai.operation.name",
+        "fieldContext": "attribute",
+        "config": {
+          "sources": [
+            {
+              "key": "openinference.span.kind",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 30
+            },
+            {
+              "key": "llm.request.type",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 20
+            },
+            {
+              "key": "traceloop.span.kind",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 10
+            }
+          ]
+        }
+      },
       {
         "name": "gen_ai.usage.input_tokens",
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "input_tokens",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 60
+            },
             {
               "key": "gen_ai.usage.prompt_tokens",
               "context": "attribute",
@@ -157,6 +204,12 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "output_tokens",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 60
+            },
             {
               "key": "gen_ai.usage.completion_tokens",
               "context": "attribute",
@@ -195,6 +248,24 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "cache_read_tokens",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 60
+            },
+            {
+              "key": "gen_ai.usage.input_tokens.cached",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 50
+            },
+            {
+              "key": "gen_ai.usage.prompt_token_details.cached_read_tokens",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 40
+            },
             {
               "key": "gen_ai.usage.cache_read_input_tokens",
               "context": "attribute",
@@ -221,6 +292,12 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "cache_creation_tokens",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 40
+            },
             {
               "key": "gen_ai.usage.cache_write.input_tokens",
               "context": "attribute",
@@ -242,6 +319,38 @@
           ]
         }
       },
+      {
+        "name": "gen_ai.usage.reasoning.output_tokens",
+        "fieldContext": "attribute",
+        "config": {
+          "sources": [
+            {
+              "key": "gen_ai.usage.reasoning_tokens",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 40
+            },
+            {
+              "key": "gen_ai.usage.completion_token_details.reasoning_tokens",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 30
+            },
+            {
+              "key": "llm.token_count.completion_details.reasoning",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 20
+            },
+            {
+              "key": "ai.usage.reasoningTokens",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 10
+            }
+          ]
+        }
+      },
       {
         "name": "gen_ai.input.messages",
         "fieldContext": "attribute",
@@ -305,6 +414,12 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "gen_ai.client.session_id",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 30
+            },
             {
               "key": "session.id",
               "context": "attribute",
@@ -325,6 +440,30 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "llm.finish_reason",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 50
+            },
+            {
+              "key": "llm.response.stop_reason",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 40
+            },
+            {
+              "key": "llm.response.finish_reason",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 30
+            },
+            {
+              "key": "gen_ai.response.finish_reason",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 20
+            },
             {
               "key": "ai.response.finishReason",
               "context": "attribute",

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

@@ -1,5 +1,5 @@
 {
-  "version": 2,
+  "version": 3,
   "definition": {
     "name": "gen_ai.tool",
     "condition": {
@@ -15,6 +15,9 @@
         },
         {
           "value": "tool_response"
+        },
+        {
+          "value": "tool_use_id"
         }
       ],
       "resource": []
@@ -26,6 +29,18 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "gen_ai.client.tool_name",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 40
+            },
+            {
+              "key": "tool_name",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 30
+            },
             {
               "key": "tool.name",
               "context": "attribute",
@@ -46,6 +61,12 @@
         "fieldContext": "attribute",
         "config": {
           "sources": [
+            {
+              "key": "gen_ai.client.tool_use_id",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 30
+            },
             {
               "key": "tool.id",
               "context": "attribute",
@@ -138,6 +159,32 @@
             }
           ]
         }
+      },
+      {
+        "name": "gen_ai.operation.name",
+        "fieldContext": "attribute",
+        "config": {
+          "sources": [
+            {
+              "key": "openinference.span.kind",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 30
+            },
+            {
+              "key": "llm.request.type",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 20
+            },
+            {
+              "key": "traceloop.span.kind",
+              "context": "attribute",
+              "operation": "copy",
+              "priority": 10
+            }
+          ]
+        }
       }
     ]
   }

pkg/query-service/app/logparsingpipeline/collector_config.go

@@ -2,6 +2,7 @@ package logparsingpipeline
 
 import (
 	"encoding/json"
+	"slices"
 	"strings"
 	"sync"
 
@@ -24,10 +25,13 @@ var (
 )
 
 const (
-	memoryLimiterProcessor       = "memory_limiter"
-	memoryLimiterProcessorPrefix = "memory_limiter/"
-	batchProcessor               = "batch"
-	batchProcessorPrefix         = "batch/"
+	memoryLimiterProcessor        = "memory_limiter"
+	memoryLimiterProcessorPrefix  = "memory_limiter/"
+	batchProcessor                = "batch"
+	batchProcessorPrefix          = "batch/"
+	normalizerProcessor           = "signozlogsnormalizer"
+	normalizerProcessorPrefix     = "signozlogsnormalizer/"
+	normalizerBeforeUserPipelines = "before"
 )
 
 // check if the processors already exist
@@ -90,9 +94,14 @@ func getOtelPipelineFromConfig(config map[string]interface{}) (*otelPipeline, er
 //  2. signoz user-pipeline processors (in the order given by signozPipelineProcessorNames)
 //  3. custom processors (non-signoz, non-memory_limiter, non-batch processors from the current config)
 //  4. batch processors (any processor named "batch" or "batch/<id>") and anything after them
+//
+// If a signozlogsnormalizer processor is present, the order above is skipped: the first one
+// runs right after user-pipeline processors, or right before them when its config sets
+// `opamp.position: before`, and the other processors keep their places.
 func buildCollectorPipelineProcessorsList(
 	currentCollectorProcessors []string,
 	signozPipelineProcessorNames []string,
+	processorConfigs map[string]interface{},
 ) ([]string, error) {
 	lockLogsPipelineSpec.Lock()
 	defer lockLogsPipelineSpec.Unlock()
@@ -103,6 +112,23 @@ func buildCollectorPipelineProcessorsList(
 		userPipelineSet[p] = struct{}{}
 	}
 
+	if normalizerIdx := slices.IndexFunc(currentCollectorProcessors, isNormalizerProcessor); normalizerIdx >= 0 {
+		normalizerFirst := opampPosition(processorConfigs, currentCollectorProcessors[normalizerIdx]) == normalizerBeforeUserPipelines
+		result := make([]string, 0, len(currentCollectorProcessors)+len(signozPipelineProcessorNames))
+		for idx, p := range currentCollectorProcessors {
+			if idx == normalizerIdx && !normalizerFirst {
+				result = append(result, signozPipelineProcessorNames...)
+			}
+			if _, inUserPipelines := userPipelineSet[p]; !inUserPipelines && !hasSignozPipelineProcessorPrefix(p) {
+				result = append(result, p)
+			}
+			if idx == normalizerIdx && normalizerFirst {
+				result = append(result, signozPipelineProcessorNames...)
+			}
+		}
+		return result, nil
+	}
+
 	var memoryLimiters []string
 	var customProcessors []string
 	batchIdx := -1
@@ -135,13 +161,51 @@ func buildCollectorPipelineProcessorsList(
 	return result, nil
 }
 
-func GenerateCollectorConfigWithPipelines(config []byte, pipelines []pipelinetypes.GettablePipeline) ([]byte, error) {
+func isNormalizerProcessor(name string) bool {
+	return name == normalizerProcessor || strings.HasPrefix(name, normalizerProcessorPrefix)
+}
+
+// collectorNormalizesFirst reports whether the logs pipeline of the collector config runs
+// signozlogsnormalizer ahead of user pipelines, judged by the first one as in buildCollectorPipelineProcessorsList.
+// Malformed configs report false; generateCollectorConfigWithPipelines surfaces their errors.
+func collectorNormalizesFirst(collectorConf map[string]interface{}) bool {
+	p, err := getOtelPipelineFromConfig(collectorConf)
+	if err != nil || p.Pipelines.Logs == nil {
+		return false
+	}
+	normalizerIdx := slices.IndexFunc(p.Pipelines.Logs.Processors, isNormalizerProcessor)
+	if normalizerIdx < 0 {
+		return false
+	}
+	processorConfigs, _ := collectorConf["processors"].(map[string]interface{})
+	return opampPosition(processorConfigs, p.Pipelines.Logs.Processors[normalizerIdx]) == normalizerBeforeUserPipelines
+}
+
+func opampPosition(processorConfigs map[string]interface{}, name string) string {
+	config, _ := processorConfigs[name].(map[string]interface{})
+	opamp, _ := config["opamp"].(map[string]interface{})
+	position, _ := opamp["position"].(string)
+	return position
+}
+
+func unmarshalCollectorConfig(config []byte) (map[string]interface{}, error) {
 	var collectorConf map[string]interface{}
-	err := yaml.Unmarshal([]byte(config), &collectorConf)
-	if err != nil {
+	if err := yaml.Unmarshal(config, &collectorConf); err != nil {
 		return nil, errors.WrapInvalidInputf(err, CodeCollectorConfigUnmarshalFailed, "could not unmarshal collector config")
 	}
+	return collectorConf, nil
+}
 
+func GenerateCollectorConfigWithPipelines(config []byte, pipelines []pipelinetypes.GettablePipeline) ([]byte, error) {
+	collectorConf, err := unmarshalCollectorConfig(config)
+	if err != nil {
+		return nil, err
+	}
+	return generateCollectorConfigWithPipelines(collectorConf, pipelines)
+}
+
+// generateCollectorConfigWithPipelines modifies collectorConf in place.
+func generateCollectorConfigWithPipelines(collectorConf map[string]any, pipelines []pipelinetypes.GettablePipeline) ([]byte, error) {
 	signozPipelineProcessors, signozPipelineProcNames, err := PreparePipelineProcessor(pipelines)
 	if err != nil {
 		return nil, err
@@ -180,11 +244,11 @@ func GenerateCollectorConfigWithPipelines(config []byte, pipelines []pipelinetyp
 		return nil, errors.NewInternalf(CodeCollectorConfigLogsPipelineNotFound, "logs pipeline doesn't exist")
 	}
 
-	updatedProcessorList, _ := buildCollectorPipelineProcessorsList(p.Pipelines.Logs.Processors, signozPipelineProcNames)
+	updatedProcessorList, _ := buildCollectorPipelineProcessorsList(p.Pipelines.Logs.Processors, signozPipelineProcNames, collectorConf["processors"].(map[string]any))
 	p.Pipelines.Logs.Processors = updatedProcessorList
 
 	// add the new processor to the data ( no checks required as the keys will exists)
-	collectorConf["service"].(map[string]interface{})["pipelines"].(map[string]interface{})["logs"] = p.Pipelines.Logs
+	collectorConf["service"].(map[string]any)["pipelines"].(map[string]any)["logs"] = p.Pipelines.Logs
 
 	updatedConf, err := yaml.Marshal(collectorConf)
 	if err != nil {

pkg/query-service/app/logparsingpipeline/collector_config_test.go

@@ -5,11 +5,14 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
 	"github.com/SigNoz/signoz/pkg/query-service/constants"
 	"github.com/SigNoz/signoz/pkg/query-service/model"
 	v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
 	"github.com/SigNoz/signoz/pkg/types/pipelinetypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
 	. "github.com/smartystreets/goconvey/convey"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v3"
 )
@@ -206,13 +209,74 @@ var BuildLogsPipelineTestData = []struct {
 func TestBuildLogsPipeline(t *testing.T) {
 	for _, test := range BuildLogsPipelineTestData {
 		Convey(test.Name, t, func() {
-			v, err := buildCollectorPipelineProcessorsList(test.fromCollector, test.userPipelines)
+			v, err := buildCollectorPipelineProcessorsList(test.fromCollector, test.userPipelines, nil)
 			So(err, ShouldBeNil)
 			So(v, ShouldResemble, test.finalOutput)
 		})
 	}
 }
 
+func TestBuildLogsPipelineWithNormalizer(t *testing.T) {
+	a := constants.LogsPPLPfx + "a"
+	b := constants.LogsPPLPfx + "b"
+	before := map[string]any{"opamp": map[string]any{"position": "before"}}
+	after := map[string]any{"opamp": map[string]any{"position": "after"}}
+
+	testCases := []struct {
+		name          string
+		fromCollector []string
+		userPipelines []string
+		configs       map[string]any
+		expected      []string
+	}{
+		{
+			name:          "ExistingAndStaleUserProcessorsReplaced",
+			fromCollector: []string{"memory_limiter", a, constants.LogsPPLPfx + "stale", constants.OldLogsPPLPfx + "old", "signozlogsnormalizer", "custom", "batch"},
+			userPipelines: []string{a, b},
+			expected:      []string{"memory_limiter", a, b, "signozlogsnormalizer", "custom", "batch"},
+		},
+		{
+			name:          "NamedNormalizerAfter_CustomProcessorsKeepPlace",
+			fromCollector: []string{"memory_limiter", "custom_a", "signozlogsnormalizer/x", a, "custom_b", "batch"},
+			userPipelines: []string{a},
+			configs:       map[string]any{"signozlogsnormalizer/x": after},
+			expected:      []string{"memory_limiter", "custom_a", a, "signozlogsnormalizer/x", "custom_b", "batch"},
+		},
+		{
+			name:          "FirstNormalizerAfter_SecondIgnored",
+			fromCollector: []string{"memory_limiter", "signozlogsnormalizer/first", "signozlogsnormalizer/second", "batch"},
+			userPipelines: []string{a},
+			configs:       map[string]any{"signozlogsnormalizer/first": after, "signozlogsnormalizer/second": before},
+			expected:      []string{"memory_limiter", a, "signozlogsnormalizer/first", "signozlogsnormalizer/second", "batch"},
+		},
+		{
+			name:          "FirstNormalizerBefore_SecondIgnored",
+			fromCollector: []string{"memory_limiter", "signozlogsnormalizer/first", "signozlogsnormalizer/second", "batch"},
+			userPipelines: []string{a},
+			configs:       map[string]any{"signozlogsnormalizer/first": before, "signozlogsnormalizer/second": after},
+			expected:      []string{"memory_limiter", "signozlogsnormalizer/first", a, "signozlogsnormalizer/second", "batch"},
+		},
+		{
+			name:          "NormalizerAfterBatch",
+			fromCollector: []string{"memory_limiter", "batch", "signozlogsnormalizer"},
+			userPipelines: []string{a},
+			expected:      []string{"memory_limiter", "batch", a, "signozlogsnormalizer"},
+		},
+		{
+			name:          "NoUserPipelines_StaleRemoved",
+			fromCollector: []string{"memory_limiter", a, "signozlogsnormalizer", "batch"},
+			expected:      []string{"memory_limiter", "signozlogsnormalizer", "batch"},
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			processors, err := buildCollectorPipelineProcessorsList(testCase.fromCollector, testCase.userPipelines, testCase.configs)
+			require.NoError(t, err)
+			assert.Equal(t, testCase.expected, processors)
+		})
+	}
+}
+
 func TestPipelineAliasCollisionsDontResultInDuplicateCollectorProcessors(t *testing.T) {
 	require := require.New(t)
 
@@ -437,3 +501,61 @@ func TestPipeCharInAliasDoesntBreakCollectorConfig(t *testing.T) {
 		}, result[0].Attributes_string,
 	)
 }
+
+func TestRecommendAgentConfigWithNormalizerProcessorAndJSONBody(t *testing.T) {
+	userPipeline := makeTestAddAttributePipeline()
+	controller := &LogParsingPipelineController{
+		fl: flaggertest.WithUseJSONBody(t, true),
+		GetIntegrationPipelines: func(context.Context, string) ([]pipelinetypes.GettablePipeline, error) {
+			return []pipelinetypes.GettablePipeline{userPipeline}, nil
+		},
+	}
+	normalizeOperator := CollectorConfProcessorName(getNormalizePipeline())
+	user := CollectorConfProcessorName(userPipeline)
+
+	testCases := []struct {
+		name       string
+		processors string
+		logs       string
+		expected   []any
+	}{
+		{
+			name:     "NoNormalizer_OperatorPrepended",
+			logs:     "[memory_limiter, batch]",
+			expected: []any{"memory_limiter", normalizeOperator, user, "batch"},
+		},
+		{
+			name:     "NormalizerDefault_OperatorKept",
+			logs:     "[memory_limiter, signozlogsnormalizer, batch]",
+			expected: []any{"memory_limiter", normalizeOperator, user, "signozlogsnormalizer", "batch"},
+		},
+		{
+			name:       "NormalizerBefore_OperatorDropped",
+			processors: "signozlogsnormalizer:\n    opamp:\n      position: before\n",
+			logs:       "[memory_limiter, signozlogsnormalizer, batch]",
+			expected:   []any{"memory_limiter", "signozlogsnormalizer", user, "batch"},
+		},
+		{
+			name:       "NormalizerAfter_OperatorKept",
+			processors: "signozlogsnormalizer:\n    opamp:\n      position: after\n",
+			logs:       "[memory_limiter, signozlogsnormalizer, batch]",
+			expected:   []any{"memory_limiter", normalizeOperator, user, "signozlogsnormalizer", "batch"},
+		},
+	}
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			conf := fmt.Sprintf(
+				"processors:\n  memory_limiter: {}\n  batch: {}\n  %s\nservice:\n  pipelines:\n    logs:\n      receivers: [otlp]\n      processors: %s\n      exporters: [clickhouselogsexporter]\n",
+				testCase.processors, testCase.logs,
+			)
+
+			recommended, _, err := controller.RecommendAgentConfig(valuer.GenerateUUID(), []byte(conf), nil)
+			require.NoError(t, err)
+
+			var recommendedConf map[string]any
+			require.NoError(t, yaml.Unmarshal(recommended, &recommendedConf))
+			logs := recommendedConf["service"].(map[string]any)["pipelines"].(map[string]any)["logs"].(map[string]any)
+			assert.Equal(t, testCase.expected, logs["processors"])
+		})
+	}
+}

pkg/query-service/app/logparsingpipeline/controller.go

@@ -220,26 +220,17 @@ func (ic *LogParsingPipelineController) ValidatePipelines(ctx context.Context,
 	return err
 }
 
-// withNormalizePipeline places normalize where the read path dictates. Ahead of user pipelines
-// when queries run on body_v2 (use_json_body), so operators see the body the explorer shows.
-// After them when dual ingestion alone writes body_v2, so operators keep seeing the raw body
-// users still query. Absent when neither flag is on.
-func (ic *LogParsingPipelineController) withNormalizePipeline(ctx context.Context, orgID valuer.UUID, pipelines []pipelinetypes.GettablePipeline) []pipelinetypes.GettablePipeline {
-	evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
-	dualIngestion := ic.fl.BooleanOrEmpty(ctx, flagger.FeatureJSONBodyDualIngestion, evalCtx)
-	switch {
-	case ic.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, evalCtx):
-		return append([]pipelinetypes.GettablePipeline{getNormalizePipeline(dualIngestion)}, pipelines...)
-	case dualIngestion:
-		return append(slices.Clone(pipelines), getNormalizePipeline(true))
-	default:
+// withNormalizePipeline places normalize ahead of user pipelines when queries run on body_v2
+// (use_json_body), so operators see the body the explorer shows. Absent when the flag is off
+// or collectorNormalizesFirst.
+func (ic *LogParsingPipelineController) withNormalizePipeline(ctx context.Context, orgID valuer.UUID, pipelines []pipelinetypes.GettablePipeline, collectorNormalizesFirst bool) []pipelinetypes.GettablePipeline {
+	if collectorNormalizesFirst || !ic.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
 		return pipelines
 	}
+	return append([]pipelinetypes.GettablePipeline{getNormalizePipeline()}, pipelines...)
 }
 
-// stashOriginalBody makes normalize carry the pre-normalization body in an internal attribute
-// for the exporter to restore into the legacy body column.
-func getNormalizePipeline(stashOriginalBody bool) pipelinetypes.GettablePipeline {
+func getNormalizePipeline() pipelinetypes.GettablePipeline {
 	return pipelinetypes.GettablePipeline{
 		StoreablePipeline: pipelinetypes.StoreablePipeline{
 			Name:    "Default Pipeline - PreProcessing Body",
@@ -258,11 +249,10 @@ func getNormalizePipeline(stashOriginalBody bool) pipelinetypes.GettablePipeline
 		},
 		Config: []pipelinetypes.PipelineOperator{
 			{
-				ID:                    uuid.NewString(),
-				Type:                  "normalize",
-				Enabled:               true,
-				If:                    "body != nil",
-				JSONBodyDualIngestion: stashOriginalBody,
+				ID:      uuid.NewString(),
+				Type:    "normalize",
+				Enabled: true,
+				If:      "body != nil",
 			},
 		},
 	}
@@ -371,11 +361,8 @@ func (ic *LogParsingPipelineController) PreviewLogsPipelines(
 	}
 
 	// The collector gets the same pipeline prepended over opamp; see RecommendAgentConfig.
-	// Under dual ingestion alone it runs after user operators and only feeds body_v2, which
-	// the explorer does not show yet, so the preview leaves it out. The original-body stash
-	// is left off: the preview has no exporter to restore and strip it.
 	if ic.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
-		pipelines = append([]pipelinetypes.GettablePipeline{getNormalizePipeline(false)}, pipelines...)
+		pipelines = append([]pipelinetypes.GettablePipeline{getNormalizePipeline()}, pipelines...)
 	}
 
 	result, collectorLogs, err := SimulatePipelinesProcessing(ctx, pipelines, request.Logs)
@@ -396,14 +383,14 @@ func (pc *LogParsingPipelineController) AgentFeatureType() agentConf.AgentFeatur
 
 // Implements agentConf.AgentFeature interface.
 // RecommendAgentConfig generates the collector config to be sent to agents.
-// The normalize pipeline (when use_json_body or json_body_dual_ingestion is on) is placed
+// The normalize pipeline (when use_json_body is on) is placed
 // here, after rawPipelineData is serialized. So it is only present in the config sent to
 // the collector and never persisted to the database as part of the user's pipeline list.
 //
 // NOTE: The configId sent to agents is derived from the pipeline version number
 // (e.g. "LogPipelines:5"), not the YAML content. If server-side logic changes
-// the generated YAML without bumping the version (e.g. toggling the use_json_body or
-// json_body_dual_ingestion flags or updating operator IfExpressions), agents that already
+// the generated YAML without bumping the version (e.g. toggling the use_json_body
+// flag or updating operator IfExpressions), agents that already
 // applied that version will not re-apply the new config. In such cases, users must save a
 // new pipeline version via the API to force agents to pick up the change.
 func (pc *LogParsingPipelineController) RecommendAgentConfig(
@@ -431,10 +418,15 @@ func (pc *LogParsingPipelineController) RecommendAgentConfig(
 		return nil, "", err
 	}
 
-	// normalize is only for sending to the collector, never persisted
-	enrichedPipelines = pc.withNormalizePipeline(ctx, orgId, enrichedPipelines)
+	collectorConf, err := unmarshalCollectorConfig(currentConfYaml)
+	if err != nil {
+		return nil, "", err
+	}
 
-	updatedConf, err := GenerateCollectorConfigWithPipelines(currentConfYaml, enrichedPipelines)
+	// normalize is only for sending to the collector, never persisted
+	enrichedPipelines = pc.withNormalizePipeline(ctx, orgId, enrichedPipelines, collectorNormalizesFirst(collectorConf))
+
+	updatedConf, err := generateCollectorConfigWithPipelines(collectorConf, enrichedPipelines)
 	if err != nil {
 		return nil, "", err
 	}

pkg/types/pipelinetypes/pipeline.go

@@ -122,9 +122,6 @@ type PipelineOperator struct {
 	EnablePaths        bool   `json:"enable_paths,omitempty" yaml:"enable_paths,omitempty"`
 	PathPrefix         string `json:"path_prefix,omitempty" yaml:"path_prefix,omitempty"`
 
-	// normalize fields, set by the server only
-	JSONBodyDualIngestion bool `json:"-" yaml:"json_body_dual_ingestion,omitempty"`
-
 	// Used in Severity Parsing and JSON Flattening mapping
 	Mapping map[string][]string `json:"mapping,omitempty" yaml:"mapping,omitempty"`
 	// severity parser fields

tests/integration/tests/alertmanager/09_googlechat.py

@@ -22,7 +22,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 METRICS_DATA = "ruler/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
 METRICS_RULE = "ruler/test_scenarios/threshold_above_at_least_once/rule.json"
 LOGS_DATA = "ruler/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"

tests/integration/tests/alertmanager/11_incidentio.py

@@ -23,7 +23,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 METRICS_DATA = "ruler/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
 METRICS_RULE = "ruler/test_scenarios/threshold_above_at_least_once/rule.json"
 LOGS_DATA = "ruler/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"

tests/integration/tests/alertmanager/13_jira.py

@@ -24,7 +24,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 METRICS_DATA = "ruler/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
 METRICS_RULE = "ruler/test_scenarios/threshold_above_at_least_once/rule.json"
 LOGS_DATA = "ruler/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"

tests/integration/tests/alertmanager/14_jira_test_channel.py

@@ -30,7 +30,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 # channel test (POST /api/v1/channels/test) drives the notifier once, synchronously,
 # with a hardcoded firing test alert and no retry. The search stub decides which
 # branch runs (create / update / reopen), so the whole issue lifecycle is

tests/integration/tests/alertmanager/15_jsmops.py

@@ -25,7 +25,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 METRICS_DATA = "ruler/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
 METRICS_RULE = "ruler/test_scenarios/threshold_above_at_least_once/rule.json"
 LOGS_DATA = "ruler/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"

tests/integration/tests/alertmanager/16_jsmops_test_channel.py

@@ -24,7 +24,6 @@ from fixtures.notification_channel import (
     wait_for_alertmanager_sync,
 )
 
-
 # channel test (POST /api/v1/channels/test) drives the notifier once, synchronously,
 # with a hardcoded firing test alert and no retry: create alert on the JSM Ops
 # gateway, then append a timeline note. Default-template events + retry are in

tests/integration/tests/spanmapper/02_default_groups.py

@@ -41,7 +41,7 @@ def test_default_groups_are_seeded_and_shipped_items_are_toggle_only(
         assert groups[name]["version"] >= 1
         assert groups[name]["createdBy"] == "signoz"
     llm = groups["gen_ai.llm"]
-    assert llm["condition"]["attributes"] == [{"value": "model", "enabled": True, "origin": "system"}]
+    assert llm["condition"]["attributes"] == [{"value": "model", "enabled": True, "origin": "system"}, {"value": "llm.input_messages", "enabled": True, "origin": "system"}]
 
     list_mappers = requests.get(signoz.self.host_configs["8080"].get(f"{GROUPS_PATH}/{llm['id']}/span_mappers"), timeout=10, headers=headers)
     assert list_mappers.status_code == HTTPStatus.OK
@@ -130,6 +130,7 @@ def test_default_groups_are_seeded_and_shipped_items_are_toggle_only(
     llm_after = {g["name"]: g for g in list_groups.json()["data"]["items"]}["gen_ai.llm"]
     assert llm_after["condition"]["attributes"] == [
         {"value": "model", "enabled": False, "origin": "system"},
+        {"value": "llm.input_messages", "enabled": True, "origin": "system"},
         {"value": "gen_ai.request.model", "enabled": True, "origin": "user"},
     ]
     assert llm_after["origin"] == "system"

tests/integration/tests/tracedetail/02_thread.py

@@ -34,7 +34,14 @@ def test_thread_returns_message_spans_in_order(
                 timestamp=now - timedelta(seconds=8), trace_id=trace_id, span_id=first_llm_id, parent_span_id=root_id, name="chat gpt-4o", resources=resources, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": first_input, "gen_ai.output.messages": first_output}, attribute_write_mode="json_only"
             ),
             Traces(
-                timestamp=now - timedelta(seconds=6), trace_id=trace_id, span_id=tool_id, parent_span_id=root_id, name="execute_tool get_weather", resources=resources, attributes={"gen_ai.tool.name": "get_weather", "gen_ai.tool.call.id": "call_1", "gen_ai.tool.call.arguments": json.dumps({"city": "Bangalore"}), "gen_ai.tool.call.result": "sunny"}, attribute_write_mode="json_only"
+                timestamp=now - timedelta(seconds=6),
+                trace_id=trace_id,
+                span_id=tool_id,
+                parent_span_id=root_id,
+                name="execute_tool get_weather",
+                resources=resources,
+                attributes={"gen_ai.tool.name": "get_weather", "gen_ai.tool.call.id": "call_1", "gen_ai.tool.call.arguments": json.dumps({"city": "Bangalore"}), "gen_ai.tool.call.result": "sunny"},
+                attribute_write_mode="json_only",
             ),
             Traces(timestamp=now - timedelta(seconds=5), trace_id=trace_id, span_id=agent_id, parent_span_id=root_id, name="invoke_agent planner", resources=resources, attributes={"gen_ai.agent.name": "planner"}, attribute_write_mode="json_only"),
             Traces(timestamp=now - timedelta(seconds=4), trace_id=trace_id, span_id=second_llm_id, parent_span_id=root_id, name="chat gpt-4o", resources=resources, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": second_input}, attribute_write_mode="json_only"),