[WIP] Fix Northbound OTel propagation and streaming span lifetime

backend/apps/northbound_base_app.py

@@ -96,7 +96,7 @@ async def northbound_lifespan(_app):
     description="Northbound APIs for partners",
     version="1.0.0",
     cors_methods=["GET", "POST", "PUT", "DELETE"],
-    enable_monitoring=False,  # Disable monitoring if not needed
+    enable_monitoring=True,
     lifespan=northbound_lifespan,
 )
 if hasattr(northbound_app, "state"):

backend/management/services/agent/runtime_client.py

@@ -1,7 +1,8 @@
 # Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
 """HTTP forwarding from northbound APIs to the runtime service."""
 
-from typing import AsyncIterator, Literal
+from collections.abc import AsyncIterator
+from typing import Literal
 
 import httpx
 from fastapi.responses import StreamingResponse
@@ -13,9 +14,11 @@
     RuntimeUpstreamError,
 )
 from consts.model import AgentRequest
-from management.services.permission.identity.authentication_helpers import generate_internal_runtime_jwt
 from infrastructure.http.client import create_httpx_client
-
+from management.services.permission.identity.authentication_helpers import (
+    generate_internal_runtime_jwt,
+)
+from nexent.monitor.monitoring import get_monitoring_manager
 
 _HOP_BY_HOP_HEADERS = {
     "connection",
@@ -53,7 +56,7 @@ def _authorization_headers(user_id: str, tenant_id: str) -> dict[str, str]:
         raise RuntimeServiceUnavailableError(
             "Internal runtime authentication is not configured"
         ) from exc
-    return {"Authorization": f"Bearer {token}"}
+    return {"Authorization": f"Bearer {token}", **get_monitoring_manager().get_trace_headers()}
 
 
 def dispatch_agent_evaluation_run(

deploy/docker/compose/docker-compose.prod.yml

@@ -191,6 +191,7 @@ services:
       UMASK: "0022"
     env_file:
       - ../../env/.env
+      - ../../env/monitoring.env
     user: root
     depends_on:
       nexent-elasticsearch:

deploy/docker/compose/docker-compose.yml

@@ -213,6 +213,7 @@ services:
       UMASK: "0022"
     env_file:
       - ../../env/.env
+      - ../../env/monitoring.env
     user: root
     depends_on:
       nexent-elasticsearch:

deploy/env/monitoring.env.example

@@ -16,7 +16,7 @@ OTEL_EXPORTER_OTLP_X_API_KEY=
 OTEL_EXPORTER_OTLP_LANGFUSE_INGESTION_VERSION=
 OTEL_EXPORTER_OTLP_METRICS_ENABLED=true
 MONITORING_INSTRUMENT_REQUESTS=false
-MONITORING_FASTAPI_INCLUDED_URLS=/agent/run,/conversation/generate_title,/nb/v1/generate_title
+MONITORING_FASTAPI_INCLUDED_URLS=/agent/run,/conversation/generate_title,/nb/,/agent/internal/northbound/
 MONITORING_FASTAPI_EXCLUDED_URLS=
 MONITORING_FASTAPI_EXCLUDE_SPANS=receive,send
 # Comma-separated, case-sensitive exact names; whitespace and empty entries are ignored.

deploy/k8s/helm/nexent/charts/nexent-northbound/templates/deployment.yaml

@@ -51,6 +51,8 @@ spec:
                 name: nexent-config
             - secretRef:
                 name: nexent-env
+            - secretRef:
+                name: nexent-monitoring-env
             - secretRef:
                 name: nexent-secrets
           env:

sdk/nexent/monitor/monitoring.py

@@ -29,7 +29,7 @@
     from opentelemetry.sdk.trace import TracerProvider
     from opentelemetry.instrumentation.requests import RequestsInstrumentor
     from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
-    from opentelemetry import trace, metrics
+    from opentelemetry import trace, metrics, propagate
     from opentelemetry.sdk.resources import Resource
     OPENTELEMETRY_AVAILABLE = True
 except ImportError:
@@ -731,6 +731,18 @@ def tracer(self):
         """Get the tracer instance."""
         return self._tracer
 
+    def get_trace_headers(self) -> Dict[str, str]:
+        """Propagate only W3C trace context across internal HTTP boundaries."""
+        if not self.is_enabled:
+            return {}
+        carrier: Dict[str, str] = {}
+        try:
+            propagate.inject(carrier)
+        except Exception:
+            logger.debug("Unable to inject telemetry context", exc_info=True)
+            return {}
+        return {key: value for key, value in carrier.items() if key in {"traceparent", "tracestate"}}
+
     def setup_fastapi_app(self, app) -> bool:
         """Setup monitoring for a FastAPI application."""
         try:

test-e2e/cases/NB-OTEL-001/case.yaml

@@ -0,0 +1,87 @@
+schema_version: '1.0'
+stage: D3
+module: V5-BASELINE
+case:
+  case_id: NB-OTEL-001
+  feature_id: F-146
+  business_rule_ids:
+  - F-146-RULE-03
+  title: Northbound to Runtime complete real-model response trace
+  type: AGENT-IT
+  priority: P0
+  status: active
+  automation: automated
+  objective: Prove actual W3C cross-service trace and complete HTTP response with
+    real model and Phoenix.
+  preconditions:
+  - Northbound/Runtime run tested source with telemetry and both routes included.
+  - Task-local NEXENT_TEST_CREDENTIALS contains Northbound URL/key, real Agent and
+    skill marker, Phoenix URL/key/project.
+  test_data:
+    variants:
+    - real_model_stream
+    - unauthenticated
+  steps:
+  - order: 1
+    action: POST chat with unique W3C traceparent; consume complete SSE from real
+      model and bound skill.
+    expected: HTTP 200, actual skill marker, no error event.
+  - order: 2
+    action: Read trace from authenticated Phoenix and inspect Northbound/Runtime HTTP,
+      Agent, model and tool spans.
+    expected: Shared trace and valid ancestry; ended Northbound span encloses normal
+      Runtime/model/tool execution; actual model input and tokens visible.
+  - order: 3
+    action: POST chat without Authorization with another W3C traceparent and inspect
+      Phoenix.
+    expected: HTTP 401, ended Northbound span, no downstream invocation.
+  - order: 4
+    action: Close another real request after response headers without consuming body;
+      inspect ended HTTP trace and poll actual persisted conversation.
+    expected: Northbound HTTP span ends; original persisted run completes exactly
+      one user/assistant pair with actual skill marker, without stop or resubmission.
+  expected_results:
+  - Actual final answer agrees with real tool/model execution.
+  - Normal Northbound HTTP duration encloses Runtime completion with shared trace
+    and ancestry.
+  - Unauthorized HTTP is traced without downstream execution.
+  - Disconnected HTTP span ends while the existing persisted-run continuation policy
+    remains unchanged.
+  forbidden_side_effects:
+  - Do not record Authorization or API keys in span attributes.
+  - Do not replace real model or product collaborators.
+  integration_boundary:
+  - Actual Northbound HTTP -> Runtime -> real Agent/model/tool -> Phoenix; no mocks.
+  observations:
+  - Sanitized trace IDs, ancestry, times, token usage and HTTP status.
+  cleanup:
+  - Retain task-owned conversations for trace review; preserve user data.
+  acceptance:
+  - id: AC-STREAM
+    description: Real complete streaming and cross-service trace.
+    steps:
+    - 1
+    - 2
+    expected_results:
+    - 1
+    - 2
+    forbidden_side_effects:
+    - 1
+    - 2
+  - id: AC-AUTH
+    description: Ended rejection span with no downstream execution.
+    steps:
+    - 3
+    expected_results:
+    - 3
+    forbidden_side_effects:
+    - 1
+  - id: AC-DISCONNECT
+    description: HTTP cancellation ends its span while persisted Agent work continues.
+    steps:
+    - 4
+    expected_results:
+    - 4
+    forbidden_side_effects:
+    - 1
+    - 2

test-e2e/cases/NB-OTEL-001/execution.yaml

@@ -0,0 +1,14 @@
+schema_version: '1.0'
+case_id: NB-OTEL-001
+implementations:
+- framework: pytest
+  file: test.py
+  selector: NB-OTEL-001
+acceptance_bindings:
+  AC-STREAM:
+  - test_nb_otel_001_real_stream_and_trace_ancestry
+  AC-AUTH:
+  - test_nb_otel_001_unauthorized_trace_without_runtime
+  AC-DISCONNECT:
+  - test_nb_otel_001_disconnect_ends_http_but_keeps_persisted_run
+timeout_seconds: 360

test-e2e/cases/NB-OTEL-001/test.py

@@ -0,0 +1,156 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""NB-OTEL-001: real Northbound/Runtime/model/tool trace via Phoenix."""
+
+import json
+import os
+import time
+from pathlib import Path
+from urllib.parse import quote
+from uuid import uuid4
+
+import pytest
+import requests
+
+
+class PrivateConfig(dict):
+    """Keep test failure introspection from revealing credentials."""
+
+    def __repr__(self):
+        return '<private telemetry fixture>'
+
+
+class SecretHeader(str):
+    def __repr__(self):
+        return '<redacted authorization>'
+
+
+@pytest.fixture(scope='module')
+def config():
+    cfg = json.loads(Path(os.environ['NEXENT_TEST_CREDENTIALS']).read_text())
+    required = ['northbound_url', 'northbound_api_key', 'agent_name', 'skill_name',
+                'skill_marker', 'phoenix_url', 'phoenix_api_key', 'phoenix_project']
+    assert all(cfg.get(key) for key in required), 'Incomplete private telemetry fixture'
+    return PrivateConfig(cfg)
+
+
+def read_trace(cfg, trace_id, expected_route):
+    url = (cfg['phoenix_url'].rstrip('/') + '/v1/projects/'
+           + quote(cfg['phoenix_project'], safe='') + '/spans')
+    deadline = time.monotonic() + 45
+    while True:
+        response = requests.get(url, params={'trace_id': trace_id, 'limit': 1000},
+                                headers={'Authorization': SecretHeader('Bearer ' + cfg['phoenix_api_key'])}, timeout=15)
+        if response.status_code == 404 and time.monotonic() < deadline:
+            time.sleep(1)
+            continue
+        assert response.status_code == 200, 'Authenticated Phoenix span query failed'
+        spans = response.json()['data']
+        if any(expected_route in str(s['attributes'].get('http.route', '')) for s in spans):
+            break
+        assert time.monotonic() < deadline, 'Completed HTTP trace was not exported'
+        time.sleep(1)
+    serialized = json.dumps(spans)
+    assert cfg['northbound_api_key'] not in serialized and cfg['phoenix_api_key'] not in serialized
+    assert all(s['context']['trace_id'] == trace_id for s in spans)
+    assert all(s.get('end_time') for s in spans)
+    artifact_dir = os.environ.get('NEXENT_TEST_ARTIFACT_DIR') or os.environ.get('RESULT_DIR')
+    assert artifact_dir, 'Runner must provide a task-local artifact directory'
+    evidence = Path(artifact_dir) / 'runtime/northbound-otel'
+    evidence.mkdir(parents=True, exist_ok=True)
+    safe = [{'name': s['name'], 'context': s['context'], 'parent_id': s.get('parent_id'),
+             'start_time': s['start_time'], 'end_time': s['end_time'],
+             'status_code': s.get('status_code'), 'route': s['attributes'].get('http.route')}
+            for s in spans]
+    (evidence/f'{trace_id}.json').write_text(json.dumps(safe, indent=2))
+    return spans
+
+
+def chat(cfg, *, authenticated):
+    trace_id = uuid4().hex
+    headers = {'traceparent': '00-' + trace_id + '-1111111111111111-01'}
+    if authenticated:
+        headers['Authorization'] = SecretHeader('Bearer ' + cfg['northbound_api_key'])
+    body = {'agent_name': cfg['agent_name'], 'query': 'Read the bound skill ' + cfg['skill_name']
+            + ' exactly once and report the actual guide marker. Do not invent tool results.'}
+    response = requests.post(cfg['northbound_url'].rstrip('/') + '/nb/v1/chat/run',
+                             headers=headers, json=body, stream=True, timeout=180)
+    with response:
+        content = b''.join(response.iter_content(chunk_size=None)).decode()
+    return trace_id, response, content
+
+
+def test_nb_otel_001_real_stream_and_trace_ancestry(config):
+    trace_id, response, content = chat(config, authenticated=True)
+    assert response.status_code == 200
+    events = [json.loads(line[5:].strip()) for line in content.splitlines()
+              if line.startswith('data:') and line[5:].strip() != '[DONE]']
+    assert not any(event.get('type') == 'error' for event in events)
+    finals = ''.join(str(e.get('content', '')) for e in events if e.get('type') == 'final_answer')
+    assert config['skill_marker'] in finals
+    spans = read_trace(config, trace_id, '/nb/v1/chat/run')
+    north = [s for s in spans if s['attributes'].get('http.route', '').endswith('/nb/v1/chat/run')]
+    runtime = [s for s in spans if '/agent/internal/northbound/run' in s['attributes'].get('http.route', '')]
+    assert len(north) == 1 and len(runtime) == 1
+    root = north[0]
+    assert root['parent_id'] == '1111111111111111'
+    assert runtime[0]['parent_id'] == root['context']['span_id']
+    assert root['start_time'] <= runtime[0]['start_time'] <= runtime[0]['end_time'] <= root['end_time']
+    models = [s for s in spans if s['attributes'].get('llm.model_name')]
+    tools = [s for s in spans if s['name'] == 'agent.tool.read_skill_md']
+    agents = [s for s in spans if s['name'] == 'agent.run']
+    assert models and tools and agents
+    by_id = {s['context']['span_id']: s for s in spans}
+    for span in models + tools + agents:
+        assert root['start_time'] <= span['start_time'] <= span['end_time'] <= root['end_time']
+        cursor = span
+        for _ in range(len(spans)):
+            if cursor['context']['span_id'] == root['context']['span_id']:
+                break
+            assert cursor.get('parent_id') in by_id, 'Broken trace ancestry'
+            cursor = by_id[cursor['parent_id']]
+        else:
+            pytest.fail('Trace parent cycle')
+    assert all(s['attributes'].get('input.truncated') is False for s in models)
+    assert all(s['attributes'].get('llm.token_count.total', 0) > 0 for s in models)
+    assert any(config['skill_marker'] in str(s['attributes'].get('input.value', '')) for s in models)
+
+
+def test_nb_otel_001_unauthorized_trace_without_runtime(config):
+    trace_id, response, _content = chat(config, authenticated=False)
+    assert response.status_code == 401
+    spans = read_trace(config, trace_id, '/nb/v1/chat/run')
+    north = [s for s in spans if s['attributes'].get('http.route', '').endswith('/nb/v1/chat/run')]
+    assert len(north) == 1
+    attrs = north[0]['attributes']
+    assert attrs.get('http.status_code', attrs.get('http.response.status_code')) == 401
+    assert not any(s['name'].startswith('agent.') or s['attributes'].get('llm.model_name')
+                   or '/agent/internal/' in s['attributes'].get('http.route', '') for s in spans)
+
+
+pytestmark = [pytest.mark.case_id('NB-OTEL-001'), pytest.mark.stage('D3')]
+
+
+def test_nb_otel_001_disconnect_ends_http_but_keeps_persisted_run(config):
+    trace_id = uuid4().hex
+    headers = {'Authorization': SecretHeader('Bearer ' + config['northbound_api_key']),
+               'traceparent': '00-' + trace_id + '-1111111111111111-01'}
+    url = config['northbound_url'].rstrip('/')
+    with requests.post(url + '/nb/v1/chat/run', headers=headers, json={
+        'agent_name': config['agent_name'], 'query': 'Read the bound skill ' + config['skill_name']
+        + ' exactly once and report its actual guide marker.'}, stream=True, timeout=180) as response:
+        assert response.status_code == 200
+        conversation_id = int(response.headers['conversation_id'])
+        # Close after headers without consuming body or stopping the persisted producer.
+    spans = read_trace(config, trace_id, '/nb/v1/chat/run')
+    assert len([s for s in spans if s['attributes'].get('http.route', '').endswith('/nb/v1/chat/run')]) == 1
+    deadline = time.monotonic() + 180
+    while True:
+        history = requests.get(url + f'/nb/v1/conversations/{conversation_id}', headers={'Authorization': headers['Authorization']}, timeout=15)
+        assert history.status_code == 200
+        messages = history.json()['data']['history']
+        answers = [m['content'] for m in messages if m['role'] == 'assistant']
+        if any(config['skill_marker'] in str(answer) for answer in answers):
+            break
+        assert time.monotonic() < deadline, 'Detached persisted run did not complete'
+        time.sleep(2)
+    assert len(answers) == 1 and sum(m['role'] == 'user' for m in messages) == 1