\n{source}\n"
- record = escape(json.dumps(call, ensure_ascii=False, default=str))
- return f"|```", str(model_output), re.IGNORECASE):
yield self._legacy_parse_fallback(memory_step, model_output, finish_reason)
return
if self._use_structured_outputs_internally:
code_action = json.loads(model_output)["code"]
code_action = extract_code_from_text(code_action, self.code_block_tags) or code_action
else:
- code_action = parse_code_blobs(model_output)
+ try:
+ parsed = classify_model_output(model_output, protocol="code_action")
+ except ModelOutputProtocolError:
+ parsed = ExecutableAction(parse_code_blobs(model_output))
+ if not isinstance(parsed, ExecutableAction):
+ raise ValueError("Expected an executable action")
+ code_action = parsed.code
except Exception:
yield self._legacy_parse_fallback(memory_step, model_output, finish_reason)
return
diff --git a/sdk/nexent/core/agents/output_protocol.py b/sdk/nexent/core/agents/output_protocol.py
index 9cd5c85bad..a3439d5c26 100644
--- a/sdk/nexent/core/agents/output_protocol.py
+++ b/sdk/nexent/core/agents/output_protocol.py
@@ -1,3 +1,4 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
"""Closed output protocol for Nexent CodeAgent runtimes."""
from __future__ import annotations
@@ -16,7 +17,6 @@
from .clarification import CLARIFICATION_SCHEMA_GUIDANCE, ClarificationForm
-
OutputProtocol = Literal["code_action", "final_envelope"]
@@ -39,6 +39,7 @@ class ExecutableAction:
code: str
legacy_format: bool = False
+ final_answer: str | None = None
@dataclass(frozen=True)
@@ -88,18 +89,19 @@ class ModelOutputProtocolExhaustedError(Exception):
"""Terminal failure after the runtime exhausts protocol repair attempts."""
-_RUN_RE = re.compile(r"\A```(?P[\s\S]*?)```\Z")
-_ACTION_PREAMBLE_RE = re.compile(
- r"\A(?P(?:(?:[Tt]hink|[Tt]hought|思考)[ \t]*[:：][\s\S]*?\n)?"
- r"[ \t]*(?:[Cc]ode|代码)[ \t]*[:：])\s*(?P[\s\S]*|```[\s\S]*)\Z",
-)
-_CODE_LABEL_RE = re.compile(r"(?m)^[ \t]*(?:Code|代码)[ \t]*[:：]", re.IGNORECASE)
_FINAL_ENVELOPE_RE = re.compile(r"\A(?P[\s\S]*) \Z", re.IGNORECASE)
-_TAG_RE = re.compile(r"]{0,255}>")
_MODEL_CONTROL_TOKEN_RE = re.compile(r"<\|[^<>]{1,255}\|>")
-_CODE_MARKER_RE = re.compile(r"")
-_THINK_BLOCK_RE = re.compile(r"[\s\S]*? ")
-_THINK_MARKER_RE = re.compile(r"")
+_RUNTIME_RECORD_RE = re.compile(
+ r"", re.IGNORECASE)
+_PROTOCOL_MARKER_RE = re.compile(r"|```", re.IGNORECASE)
+_MARKDOWN_RE = re.compile(
+ r"(?m)^[ \t]{0,3}(?P`{3,}|~{3,})[^\n]*\n"
+ r"[\s\S]*?(?:^[ \t]{0,3}(?P=fence)[ \t]*(?:\n|$)|\Z)"
+ r"|(?P`+)[\s\S]*?(?P=ticks)"
+)
_CODE_OPEN = ""
_CODE_CLOSE = ""
_PYTHON_DATA_TOKEN_TYPES = {
@@ -165,8 +167,9 @@ def protocol_repair_instruction(
return (
prefix + "Return optional reasoning followed by one or more complete ... blocks. "
- "Prefer one block; if multiple blocks are needed, put only whitespace between them because they execute together as one Python action. "
- "Put no text after the final block. To finish, return exactly one complete ... envelope with no content outside it."
+ "Only top-level protocol blocks execute; Markdown examples are data. Blocks execute together in order. "
+ "To finish, put one complete ... envelope at the end. "
+ "When actions precede it, the runtime executes them and requests confirmation using the actual results."
)
@@ -225,7 +228,7 @@ def _python_literal_and_comment_spans(value: str) -> list[tuple[int, int]]:
)
except (tokenize.TokenError, IndentationError, SyntaxError):
# Lexically incomplete Python cannot form a complete executable Action.
- return []
+ return spans
return spans
@@ -244,124 +247,82 @@ def _outer_code_markers(action_region: str) -> list[re.Match[str]]:
return [match for match in matches if not _position_in_spans(match.start(), spans)]
-def _validate_reasoning_prefix(
- prefix: str,
- *,
- protocol: OutputProtocol,
- logger: Any,
-) -> None:
- prefix = strip_protocol_padding(prefix)
- if not has_meaningful_visible_content(prefix):
- return
-
- # Thinking is descriptive prefix data, not an executable envelope. Preserve
- # accepted complete spans, then tolerate remaining unbalanced/nested markers.
- # Only inspect the prefix: tags inside Python data must remain untouched.
- prefix = _THINK_BLOCK_RE.sub("", prefix)
- prefix = _THINK_MARKER_RE.sub("", prefix)
- if _TAG_RE.search(prefix) or _MODEL_CONTROL_TOKEN_RE.search(prefix):
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
+def _markdown_spans(text: str) -> list[tuple[int, int]]:
+ """Identify quoted examples before interpreting protocol markers."""
+ return [(match.start(), match.end()) for match in _MARKDOWN_RE.finditer(text)
+ if not match.group(0).lower().startswith("```")]
def is_raw_final_answer_candidate(output: Any) -> bool:
- """Exclude failed protocol-marked output from the exhaustion-only fallback."""
-
+ """Keep reserved protocol records out of the exhaustion-only fallback."""
text = strip_protocol_padding(output)
return (
has_meaningful_visible_content(text)
- and not _TAG_RE.search(text)
+ and not _PROTOCOL_MARKER_RE.search(text)
and not _MODEL_CONTROL_TOKEN_RE.search(text)
- and not any(
- marker in text.lower()
- for marker in (" ExecutableAction | None:
- first_open = text.find(_CODE_OPEN)
- if first_open < 0:
- return None
-
- prefix = text[:first_open]
- _validate_reasoning_prefix(prefix, protocol=protocol, logger=logger)
- action_region = text[first_open:]
- markers = _outer_code_markers(action_region)
- if not markers or markers[0].start() != 0:
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
-
+) -> ExecutableAction | ExplicitFinalAnswer | NonterminalThought:
bodies: list[str] = []
- expected_open = True
- previous_end = 0
- body_start = 0
- for marker in markers:
- marker_text = marker.group(0)
- if expected_open:
- if marker_text != _CODE_OPEN:
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
- if bodies and has_meaningful_visible_content(action_region[previous_end : marker.start()]):
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
- body_start = marker.end()
- else:
- if marker_text != _CODE_CLOSE:
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
- body = strip_protocol_padding(action_region[body_start : marker.start()])
- if not has_meaningful_visible_content(body):
+ position = 0
+ legacy = False
+ final_answer = None
+ while True:
+ remaining = text[position:]
+ examples = _markdown_spans(remaining)
+ marker = next((candidate for candidate in _PROTOCOL_MARKER_RE.finditer(remaining)
+ if not _position_in_spans(candidate.start(), examples)), None)
+ prefix_end = marker.start() if marker else len(remaining)
+ if any(not _position_in_spans(token.start(), examples)
+ for token in _MODEL_CONTROL_TOKEN_RE.finditer(remaining[:prefix_end])):
+ _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
+ if any(not _position_in_spans(token.start(), examples)
+ for token in _RUNTIME_RECORD_RE.finditer(remaining[:prefix_end])):
+ _raise_protocol_error(ProtocolErrorReason.UNSUPPORTED_OR_TAG_ONLY_OUTPUT, protocol, logger)
+ if marker is None:
+ break
+ start = position + marker.start()
+ end = position + marker.end()
+ tag = marker.group().lower()
+ if tag == "":
+ final = _classify_final_envelope(text[start:], protocol=protocol, logger=logger)
+ final_answer = final.answer
+ break
+ if tag == "```":
+ close = text.find("```", end)
+ if close < 0:
_raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
- bodies.append(body)
- previous_end = marker.end()
- expected_open = not expected_open
-
- if not expected_open or not bodies:
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
- if has_meaningful_visible_content(action_region[previous_end:]):
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
-
- code = "\n\n".join(bodies)
- try:
- ast.parse(code)
- except (SyntaxError, ValueError, TypeError):
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
- return ExecutableAction(code=code)
-
-
-def _classify_code_action(
- text: str,
- *,
- protocol: OutputProtocol,
- logger: Any,
-) -> ExecutableAction | NonterminalThought:
- code_action = _parse_code_action(text, protocol=protocol, logger=logger)
- if code_action is not None:
- return code_action
-
- run_start = text.find("```")
- if run_start >= 0:
- _validate_reasoning_prefix(text[:run_start], protocol=protocol, logger=logger)
- run_match = _RUN_RE.fullmatch(text[run_start:]) if run_start >= 0 else None
- if run_match and text.count("```") == 1:
- return _parse_executable_action(
- run_match.group("body"),
- protocol=protocol,
- logger=logger,
- legacy_format=True,
- )
-
- if any(marker in text for marker in ("", "", "```")):
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
- if _CODE_LABEL_RE.search(text):
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
- if _TAG_RE.search(text) or _MODEL_CONTROL_TOKEN_RE.search(text):
- _raise_protocol_error(
- ProtocolErrorReason.UNSUPPORTED_OR_TAG_ONLY_OUTPUT,
- protocol,
- logger,
- )
+ bodies.append(strip_protocol_padding(text[end:close]))
+ position = close + 3
+ legacy = True
+ continue
+ if tag != _CODE_OPEN:
+ _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
+ region = text[start:]
+ markers = _outer_code_markers(region)
+ if len(markers) < 2 or markers[1].group().lower() != _CODE_CLOSE:
+ _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
+ body = strip_protocol_padding(region[markers[0].end():markers[1].start()])
+ if not has_meaningful_visible_content(body):
+ _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
+ bodies.append(body)
+ position = start + markers[1].end()
+ if bodies:
+ action = _parse_executable_action("\n\n".join(bodies), protocol=protocol, logger=logger,
+ legacy_format=legacy)
+ return ExecutableAction(action.code, action.legacy_format, final_answer)
+ if final_answer is not None:
+ return ExplicitFinalAnswer(final_answer)
+ if re.search(r"" not in text and (
- "" in text.lower()
- or " " in text.lower()
- ):
- final_start = text.lower().find("")
- if final_start >= 0:
- _validate_reasoning_prefix(text[:final_start], protocol=protocol, logger=logger)
- text = text[final_start:]
- return _classify_final_envelope(text, protocol=protocol, logger=logger)
-
- # The platform prompt explicitly requests Think:/Code: (or 思考：/代码：).
- # Preserve labeled legacy RUN actions and reject embedded examples before
- # validating the remaining action with the shared parser.
- preamble_match = _ACTION_PREAMBLE_RE.fullmatch(text)
- if preamble_match:
- preamble = preamble_match.group("preamble")
- if any(marker in preamble for marker in ("", "", "```")):
- _raise_protocol_error(ProtocolErrorReason.MALFORMED_ACTION, protocol, logger)
- text = preamble_match.group("action")
return _classify_code_action(text, protocol=protocol, logger=logger)
+
def extract_clarification_form(code: str, tool_name: str) -> ClarificationForm | None:
"""Recognize a terminal form before any part of the Python action can execute."""
try:
diff --git a/sdk/nexent/core/prompts/en/agent/agent_manager.yaml b/sdk/nexent/core/prompts/en/agent/agent_manager.yaml
index cbc246e8df..ad6e98c804 100644
--- a/sdk/nexent/core/prompts/en/agent/agent_manager.yaml
+++ b/sdk/nexent/core/prompts/en/agent/agent_manager.yaml
@@ -41,7 +41,7 @@ system_sections:
final_answer_guidance: |-
### Final Answer Requirements
- After thinking, when you are ready to answer, return exactly one `... ` envelope with the complete user-facing answer inside it. Do not wrap the envelope in ``, put content outside it, or perform another action afterward.
+ After thinking, when you are ready to answer, return exactly one `... ` envelope with the complete user-facing answer inside it. Place the envelope last. If code actions precede it, wait for runtime execution and confirm or revise the answer using actual results.
Follow the user's requested response format strictly; do not add headings, paragraphs, or fields the user did not request.
When generating the final answer, you need to follow these specifications:
diff --git a/sdk/nexent/core/prompts/en/agent/agent_worker.yaml b/sdk/nexent/core/prompts/en/agent/agent_worker.yaml
index 5c229b3be3..95cdf4c93c 100644
--- a/sdk/nexent/core/prompts/en/agent/agent_worker.yaml
+++ b/sdk/nexent/core/prompts/en/agent/agent_worker.yaml
@@ -41,7 +41,7 @@ system_sections:
final_answer_guidance: |-
### Final Answer Requirements
- After thinking, when you are ready to answer, return exactly one `... ` envelope with the complete user-facing answer inside it. Do not wrap the envelope in ``, put content outside it, or perform another action afterward.
+ After thinking, when you are ready to answer, return exactly one `... ` envelope with the complete user-facing answer inside it. Place the envelope last. If code actions precede it, wait for runtime execution and confirm or revise the answer using actual results.
Follow the user's requested response format strictly; do not add headings, paragraphs, or fields the user did not request.
When generating the final answer, you need to follow these specifications:
diff --git a/sdk/nexent/core/prompts/zh/agent/agent_manager.yaml b/sdk/nexent/core/prompts/zh/agent/agent_manager.yaml
index 60bf9098d9..c346a8a059 100644
--- a/sdk/nexent/core/prompts/zh/agent/agent_manager.yaml
+++ b/sdk/nexent/core/prompts/zh/agent/agent_manager.yaml
@@ -42,7 +42,7 @@ system_sections:
final_answer_guidance: |-
### 最终回答要求
- 思考结束后，当你认为可以回答用户问题，必须输出唯一一对 `... `，并将完整的用户可见回答放在标签内。不得在标签外输出内容，也不得继续执行动作。
+ 思考结束后，当你认为可以回答用户问题，必须输出唯一一对 `... `，并将完整的用户可见回答放在标签内。最终回答必须位于本轮末尾。若前面包含动作，先等待运行时执行，再根据实际结果确认或修正最终回答。
用户明确指定响应格式时，应严格按该格式回答；不擅自增加用户未要求的标题、段落或字段。
生成最终回答时，你需要遵循以下规范：
diff --git a/sdk/nexent/core/prompts/zh/agent/agent_worker.yaml b/sdk/nexent/core/prompts/zh/agent/agent_worker.yaml
index be297da6f2..36f450a181 100644
--- a/sdk/nexent/core/prompts/zh/agent/agent_worker.yaml
+++ b/sdk/nexent/core/prompts/zh/agent/agent_worker.yaml
@@ -41,7 +41,7 @@ system_sections:
final_answer_guidance: |-
### 最终回答要求
- 思考结束后，当你认为可以回答用户问题，必须输出唯一一对 `... `，将完整的用户可见回答放在标签内。不得在标签外输出内容，也不得继续执行动作。
+ 思考结束后，当你认为可以回答用户问题，必须输出唯一一对 `... `，将完整的用户可见回答放在标签内。最终回答必须位于本轮末尾。若前面包含动作，先等待运行时执行，再根据实际结果确认或修正最终回答。
用户明确指定响应格式时，应严格按该格式回答；不擅自增加用户未要求的标题、段落或字段。
生成最终回答时，你需要遵循以下规范：
diff --git a/sdk/nexent/core/utils/observer.py b/sdk/nexent/core/utils/observer.py
index 64c5a72bd8..0614978c79 100644
--- a/sdk/nexent/core/utils/observer.py
+++ b/sdk/nexent/core/utils/observer.py
@@ -385,6 +385,7 @@ def _reset_model_stream_state(self) -> None:
self.in_think_mode = False
self.legacy_code_fence_pending = False
self._stream_state.in_final_answer = False
+ self._stream_state.defer_action_answer = False
self._step_output().pop("answer", None)
self._step_output().pop("body", None)
self._step_output().pop("accepted_answer", None)
@@ -443,6 +444,17 @@ def add_model_new_token(self, new_token):
if self.token_buffer:
self._emit(self.current_mode, "".join(self.token_buffer))
self.token_buffer.clear()
+ from ..agents.output_protocol import (
+ ExecutableAction,
+ ModelOutputProtocolError,
+ classify_model_output,
+ )
+
+ try:
+ preceding = classify_model_output("".join(self._step_output().get("body", [])), protocol="code_action")
+ except ModelOutputProtocolError:
+ preceding = None
+ self._stream_state.defer_action_answer = isinstance(preceding, ExecutableAction)
self._stream_state.in_final_answer = True
elif marker != " ":
self.in_think_mode = not self.in_think_mode
@@ -463,6 +475,8 @@ def add_model_new_token(self, new_token):
def _emit_answer_delta(self, content: str) -> None:
"""Publish answer fragments in the same invocation and model attempt."""
+ if getattr(self._stream_state, "defer_action_answer", False):
+ return
state = self._step_output()
state.setdefault("answer", []).append(content)
self._emit(ProcessType.FINAL_ANSWER, content, metadata={"answer_phase": "delta"})
diff --git a/sdk/nexent/monitor/monitoring.py b/sdk/nexent/monitor/monitoring.py
index e2171f60fc..17d18d60a8 100644
--- a/sdk/nexent/monitor/monitoring.py
+++ b/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:
diff --git a/test-e2e/cases/NB-OTEL-001/case.yaml b/test-e2e/cases/NB-OTEL-001/case.yaml
new file mode 100644
index 0000000000..638cc2cd5b
--- /dev/null
+++ b/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
diff --git a/test-e2e/cases/NB-OTEL-001/execution.yaml b/test-e2e/cases/NB-OTEL-001/execution.yaml
new file mode 100644
index 0000000000..6abf5f87c8
--- /dev/null
+++ b/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
diff --git a/test-e2e/cases/NB-OTEL-001/test.py b/test-e2e/cases/NB-OTEL-001/test.py
new file mode 100644
index 0000000000..31a6159b91
--- /dev/null
+++ b/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 ''
+
+
+class SecretHeader(str):
+ def __repr__(self):
+ return ''
+
+
+@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
diff --git a/test-e2e/changes/bugs/NB-OTEL-20261111.yaml b/test-e2e/changes/bugs/NB-OTEL-20261111.yaml
new file mode 100644
index 0000000000..b4f82c36a1
--- /dev/null
+++ b/test-e2e/changes/bugs/NB-OTEL-20261111.yaml
@@ -0,0 +1,53 @@
+schema_version: '1.0'
+change:
+ change_id: NB-OTEL-20261111
+ change_type: bugfix
+ bug_id: NB-OTEL-20261111
+ title: Trace Northbound HTTP through complete response and Runtime
+ description: Enable ASGI instrumentation, supply deployment monitoring config and
+ propagate W3C context.
+ observed_behavior: Northbound monitoring is disabled; env/secret and Runtime traceparent
+ are missing.
+ expected_behavior: Included requests trace receipt through final send with Runtime
+ in the same trace.
+ escaped_stage: D3
+ coverage_gap:
+ type: missing_boundary
+ description: Existing proxy and instrumentation UT/formal chat tests lack final
+ send and cross-service trace assertions.
+ feature_contract_changed: true
+ affected_features:
+ existing:
+ - F-146
+ added: []
+ modified:
+ - F-146
+ retired: []
+ affected_cases:
+ existing: []
+ added:
+ - NB-OTEL-001
+ modified: []
+ retired: []
+ traditional_ut:
+ decision: add
+ behaviors:
+ - ASGI final send lifetime, errors/cancellation and context isolation
+ - W3C propagation, disabled/missing telemetry and no baggage
+ - Docker/Helm monitoring parity and URL filters
+ test_selectors:
+ - test/sdk/monitor/test_http_response_lifecycle.py
+ - test/backend/services/test_runtime_proxy_telemetry.py
+ - test/backend/deploy/test_k8s_env_injection.py
+ reason: New unit boundaries and deployed D3 provide distinct missing lifecycle
+ and ancestry evidence.
+ notes: 'SPEC ../nexent-doc/nexent/codeagent-action-history-role/changes/v285-northbound-otel.
+ User-requested current PR4328 extension. No new D1: isolated UT and real D3 suffice;
+ existing disconnect/final-answer cases are adjacent regressions. Existing NB-CHAT-DISCONNECT-001
+ and NB-CHAT-FINAL-ANSWER-001 were reviewed as unchanged wider business contracts,
+ not selected for full execution: relevant normal SSE and new-conversation disconnect
+ compatibility are exercised by NB-OTEL-001; existing/exhaustive answer-mode unit
+ suites remain adjacent regression. CI found seven stale Northbound env-injection/precedence
+ expectations in the existing Helm suite; align its monitored-service set, preserving
+ all ordering/platform/credential precedence assertions. Entire owning suite 24
+ tests passed locally.'
diff --git a/test-e2e/changes/bugs/OUTPUT-PROTOCOL-HISTORY-20261111.yaml b/test-e2e/changes/bugs/OUTPUT-PROTOCOL-HISTORY-20261111.yaml
new file mode 100644
index 0000000000..50b4c599d2
--- /dev/null
+++ b/test-e2e/changes/bugs/OUTPUT-PROTOCOL-HISTORY-20261111.yaml
@@ -0,0 +1,53 @@
+schema_version: '1.0'
+change:
+ change_id: OUTPUT-PROTOCOL-HISTORY-20261111
+ change_type: bugfix
+ bug_id: OUTPUT-PROTOCOL-HISTORY-20261111
+ title: Converge output parsing and restore runtime-only read-only history
+ description: Accept HTML reasoning and ordered top-level actions while excluding
+ Markdown examples; replay actual calls/results without model narration.
+ observed_behavior: Generic tag checks reject HTML; raw assistant narration is replayed
+ as history.
+ expected_behavior: Precise protocol scanning and one read-only execution record
+ preserve result deduplication, masking and recall.
+ escaped_stage: D3
+ coverage_gap:
+ type: missing_boundary
+ description: Existing protocol UT enforce old prose rejection and history UT require
+ model replay; strengthen accepted contracts and add actual Agent request regression.
+ feature_contract_changed: true
+ affected_features:
+ existing:
+ - TPC-001
+ - F-171
+ added: []
+ modified:
+ - TPC-001
+ retired: []
+ affected_cases:
+ existing:
+ - TPC-D3-001
+ - TPC-D4-001
+ - AGENT-HISTORY-D3-001
+ added: []
+ modified: []
+ retired: []
+ traditional_ut:
+ decision: strengthen
+ behaviors:
+ - Top-level case-insensitive actions with prose and HTML
+ - Markdown examples never execute
+ - Mixed candidate executes once before actual-result confirmation
+ - Runtime-only history preserves masking, false values and recall
+ test_selectors:
+ - test/sdk/core/agents/test_output_protocol_patch.py
+ - test/sdk/core/agents/test_output_protocol.py
+ - test/sdk/core/agents/test_execution_history_compaction.py
+ - test/sdk/core/agents/test_execution_history_review_regressions.py
+ reason: Existing UT and deployed TPC/history cases cover adjacent behavior; new
+ regression checks actual next model requests, no new formal D1 needed.
+ notes: SPEC ../nexent-doc/nexent/codeagent-action-history-role/changes/v285-protocol-history-patch/.
+ User authorized implementation and push before full VM acceptance. Traditional
+ UT and formal runtime evidence reported separately. Existing formal D3/D4 remain
+ required and must execute against the patch; new input variants verified by patch
+ acceptance script.
diff --git a/test-e2e/features/F-146/cases.md b/test-e2e/features/F-146/cases.md
index 493d605cb5..aa9c9cd8a9 100644
--- a/test-e2e/features/F-146/cases.md
+++ b/test-e2e/features/F-146/cases.md
@@ -7,3 +7,4 @@
- D3 [AGT-065](../../cases/AGT-065/case.yaml) · active
- D3 [AGT-066](../../cases/AGT-066/case.yaml) · active
- D3 [NB-CHAT-DISCONNECT-001](../../cases/NB-CHAT-DISCONNECT-001/case.yaml) · active
+- D3 [NB-OTEL-001](../../cases/NB-OTEL-001/case.yaml) · active
diff --git a/test-e2e/features/F-146/feature.yaml b/test-e2e/features/F-146/feature.yaml
index dd48a837d0..a691e31f91 100644
--- a/test-e2e/features/F-146/feature.yaml
+++ b/test-e2e/features/F-146/feature.yaml
@@ -6,12 +6,19 @@ feature:
feature_id: F-146
name: new/existing conversation、agent_name/query/model override、SSE
status: active
- description: 'Northbound API / Chat Run: new/existing conversation、agent_name/query/model override、SSE'
+ description: 'Northbound API / Chat Run: new/existing conversation、agent_name/query/model
+ override、SSE'
business_rules:
- rule_id: F-146-RULE-01
description: new/existing conversation、agent_name/query/model override、SSE
- rule_id: F-146-RULE-02
- description: Northbound chat defaults to one complete root final_answer; explicit final_answer_mode=delta retains incremental answer phases. Invalid modes return 422 before execution.
+ description: Northbound chat defaults to one complete root final_answer; explicit
+ final_answer_mode=delta retains incremental answer phases. Invalid modes return
+ 422 before execution.
+ - rule_id: F-146-RULE-03
+ description: When telemetry is enabled and routes included, Northbound server
+ spans last through final ASGI send and propagate W3C context to Runtime/Agent;
+ errors or HTTP disconnect close the span without changing persisted-run continuation.
required_test_stages:
- D2
- D3
diff --git a/test-e2e/features/TPC-001/feature.yaml b/test-e2e/features/TPC-001/feature.yaml
index e01b1981f7..1dea4168e5 100644
--- a/test-e2e/features/TPC-001/feature.yaml
+++ b/test-e2e/features/TPC-001/feature.yaml
@@ -10,8 +10,10 @@ feature:
answers.
business_rules:
- rule_id: TPC-RULE-001
- description: Valid actions tolerate incomplete, repeated and nested thinking prefixes
- and narration; only the Python body executes.
+ description: Top-level case-insensitive code actions tolerate ordinary HTML and
+ prose; quoted Markdown examples are data. Ordered actions execute together,
+ with an optional unique final envelope last and actual-result confirmation after
+ execution.
- rule_id: TPC-RULE-002
description: Failed protocol-marked output cannot become a raw final at exhaustion;
valid explicit completion retains its envelope boundary.
diff --git a/test/assets/sdk_prompt_migration_hashes.json b/test/assets/sdk_prompt_migration_hashes.json
index 95eca1437b..560fdf5b3b 100644
--- a/test/assets/sdk_prompt_migration_hashes.json
+++ b/test/assets/sdk_prompt_migration_hashes.json
@@ -1,6 +1,6 @@
{
- "en/agent/agent_manager.yaml": "2b47cf0ba4f07147d17a857079c9ea2e56e279c904d6fb33144926ca11f18ae3",
- "en/agent/agent_worker.yaml": "019644ad106d85a6c9c40cb184c42a788e32ba7791cd1d6729439b486901c7fb",
+ "en/agent/agent_manager.yaml": "c9aac0bbdde40adbea6eb3f79fb0fd0d4b205414c591fbd6a70728e8e0b7d64c",
+ "en/agent/agent_worker.yaml": "7d38affc5d8bf51eda4aba2290123bd5055509b16c6b8b26d54ed3ae7060ae4c",
"en/agent/answer_verifier.yaml": "dde165da09c8ce985367e1ca8400393ffa2d7c5fee8eb4ef871b3fc1db1f3d38",
"en/agent/automation_tool_policy.yaml": "57faff03c80b7226d13b2e62ea0c4ea36c089ba3df2b1c10885512433251c67f",
"en/agent/context_sections.yaml": "617be0b527ae409ebcc3a044ed815c3ff0b9ff124204d2917575fdd17eefa3e1",
@@ -35,8 +35,8 @@
"en/tool/analyze_file.yaml": "8d6b93590b13d225f88cc7c2e72dfc5df16f1d807257b8cd4e5eb924e630c6b9",
"en/tool/analyze_image.yaml": "2dfd8a05445741de56c58f55a98373baaee229b34f76b0ea3d879ea0b4b439bc",
"en/tool/analyze_video.yaml": "c63b82a6fc4d507bbd698fd9923342a1203c684554fad90c35132768e8230dcf",
- "zh/agent/agent_manager.yaml": "4d1cd938cf17d406e7f91a1f2398d0e9a7b52c18de65bd1d6234f5f432f1def0",
- "zh/agent/agent_worker.yaml": "494d042e535615d8c7debf021d6ccc84149bc69303389ae2d1076c08a0fb7b2d",
+ "zh/agent/agent_manager.yaml": "9f4a291481a6bcbf8f97bda757e4fa670429ae338d945b3c8f4790cd8bc8e7a6",
+ "zh/agent/agent_worker.yaml": "6ad7ddb9699d247af94ab0e0a02adf0245372e19df2eb1b25836efe0c24a8227",
"zh/agent/answer_verifier.yaml": "98b40d8e7a1369fccf35c244fb14613ed3aea0fa46cb659e2585f34b4b46274d",
"zh/agent/automation_tool_policy.yaml": "98d3dc237138f87f60c20e52ae758ae84d599c31735827a307b251ade28b8468",
"zh/agent/context_sections.yaml": "2f367b17eb8e03aa746775001e6e9c961311bded0f5880441bac42e558d153c8",
diff --git a/test/backend/deploy/test_k8s_env_injection.py b/test/backend/deploy/test_k8s_env_injection.py
index 47579f7745..eeb04ae449 100644
--- a/test/backend/deploy/test_k8s_env_injection.py
+++ b/test/backend/deploy/test_k8s_env_injection.py
@@ -245,7 +245,7 @@ def test_default_helm_has_empty_env_secrets_and_service_specific_references():
for v in deployment["spec"]["template"]["spec"]["containers"][0].get("envFrom", [])]
if name in BACKENDS:
expected = [None, "nexent-env"]
- if name in ("config", "runtime"):
+ if name in ("config", "runtime", "northbound"):
expected.append("nexent-monitoring-env")
assert references == expected + ["nexent-secrets"]
else:
@@ -380,8 +380,8 @@ def test_helm_composition_preserves_precedence_and_platform_settings(tmp_path, d
for deployment in deployments:
name = deployment["metadata"]["name"].removeprefix("nexent-")
container, effective = container_environment(documents, deployment)
- assert effective["SHARED"] == ("monitoring" if name in ("config", "runtime") else "root")
- assert ("MONITOR_ONLY" in effective) == (name in ("config", "runtime"))
+ assert effective["SHARED"] == ("monitoring" if name in ("config", "runtime", "northbound") else "root")
+ assert ("MONITOR_ONLY" in effective) == (name in ("config", "runtime", "northbound"))
assert effective["MINIO_ACCESS_KEY"] == "final-access"
assert effective["EMPTY"] == ""
assert effective["UNKNOWN_NEW_FEATURE"] == "enabled"
diff --git a/test/backend/services/test_runtime_proxy_telemetry.py b/test/backend/services/test_runtime_proxy_telemetry.py
new file mode 100644
index 0000000000..c69fa21000
--- /dev/null
+++ b/test/backend/services/test_runtime_proxy_telemetry.py
@@ -0,0 +1,82 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""NB-AC02/04/05: runtime propagation and deployed monitoring configuration."""
+
+import ast
+from pathlib import Path
+
+import pytest
+import yaml
+from opentelemetry import context, propagate, trace
+from opentelemetry.baggage import set_baggage
+from opentelemetry.trace import NonRecordingSpan, SpanContext, TraceFlags, TraceState
+
+from management.services.agent import runtime_client
+from nexent.monitor import monitoring
+
+
+@pytest.fixture
+def manager(monkeypatch):
+ instance = monitoring.get_monitoring_manager()
+ monkeypatch.setattr(instance, '_config', monitoring.MonitoringConfig(enable_telemetry=True))
+ monkeypatch.setattr(runtime_client, 'generate_internal_runtime_jwt', lambda *_: 'internal-only-jwt')
+ return instance
+
+
+def test_nb_ac002_internal_jwt_and_w3c_parent_survive_without_baggage(manager):
+ span_context = SpanContext(0x1234, 0x5678, True, TraceFlags(1), TraceState([('vendor', 'value')]))
+ ctx = trace.set_span_in_context(NonRecordingSpan(span_context))
+ token = context.attach(set_baggage('private', 'never-forward', context=ctx))
+ try:
+ headers = runtime_client._authorization_headers('user', 'tenant')
+ finally:
+ context.detach(token)
+ assert headers['Authorization'] == 'Bearer internal-only-jwt'
+ assert headers['traceparent'] == '00-00000000000000000000000000001234-0000000000005678-01'
+ assert headers['tracestate'] == 'vendor=value'
+ assert set(headers) == {'Authorization', 'traceparent', 'tracestate'}
+ parent = trace.get_current_span(propagate.extract(headers)).get_span_context()
+ assert parent.trace_id == span_context.trace_id and parent.span_id == span_context.span_id
+ assert manager.get_trace_headers() == {}, 'Context leaked beyond the original request'
+
+
+@pytest.mark.parametrize('condition', ['disabled', 'missing_dependencies', 'injection_error'])
+def test_nb_ac003_monitoring_failure_does_not_break_auth(manager, monkeypatch, condition):
+ if condition == 'disabled':
+ monkeypatch.setattr(manager, '_config', monitoring.MonitoringConfig(enable_telemetry=False))
+ elif condition == 'missing_dependencies':
+ monkeypatch.setattr(monitoring, 'OPENTELEMETRY_AVAILABLE', False)
+ else:
+ def fail(*_args):
+ raise RuntimeError('controlled propagation failure')
+ monkeypatch.setattr(monitoring.propagate, 'inject', fail)
+ assert runtime_client._authorization_headers('user', 'tenant') == {'Authorization': 'Bearer internal-only-jwt'}
+
+
+def test_nb_ac005_propagator_cannot_inject_authorization_or_baggage(manager, monkeypatch):
+ def inject(carrier):
+ carrier.update(traceparent='parent', Authorization='external-secret', baggage='private-data')
+ monkeypatch.setattr(monitoring.propagate, 'inject', inject)
+ assert runtime_client._authorization_headers('user', 'tenant') == {
+ 'Authorization': 'Bearer internal-only-jwt', 'traceparent': 'parent'}
+
+
+def test_nb_ac004_northbound_enables_existing_monitoring_and_deployment_config():
+ root = Path(__file__).resolve().parents[3]
+ tree = ast.parse((root/'backend/apps/northbound_base_app.py').read_text())
+ calls = [node for node in ast.walk(tree) if isinstance(node, ast.Call)
+ and isinstance(node.func, ast.Name) and node.func.id == 'create_app']
+ assert len(calls) == 1
+ assert any(kw.arg == 'enable_monitoring' and ast.literal_eval(kw.value) is True for kw in calls[0].keywords)
+ for name in ['docker-compose.yml', 'docker-compose.prod.yml']:
+ services = yaml.safe_load((root/'deploy/docker/compose'/name).read_text())['services']
+ assert '../../env/monitoring.env' in services['nexent-northbound']['env_file']
+ helm = (root/'deploy/k8s/helm/nexent/charts/nexent-northbound/templates/deployment.yaml').read_text()
+ assert 'name: nexent-monitoring-env' in helm.split('envFrom:')[1].split(' env:')[0]
+ env = (root/'deploy/env/monitoring.env.example').read_text()
+ included = next(line.split('=', 1)[1] for line in env.splitlines() if line.startswith('MONITORING_FASTAPI_INCLUDED_URLS='))
+ excluded = monitoring._build_fastapi_excluded_urls(included, '')
+ import re
+ assert not re.search(excluded, 'http://northbound:5013/api/nb/v1/chat/run')
+ assert not re.search(excluded, 'http://runtime:5014/api/agent/internal/northbound/run')
+ explicit = monitoring._build_fastapi_excluded_urls(included, '/nb/v1/chat/run')
+ assert any(re.search(pattern, 'http://northbound:5013/api/nb/v1/chat/run') for pattern in explicit.split(','))
diff --git a/test/sdk/core/agents/test_context_helper_contracts.py b/test/sdk/core/agents/test_context_helper_contracts.py
index ac91e9a52a..aff7406cb0 100644
--- a/test/sdk/core/agents/test_context_helper_contracts.py
+++ b/test/sdk/core/agents/test_context_helper_contracts.py
@@ -384,8 +384,8 @@ def test_renderer_current_action_preserves_string_tool_arguments():
text = message["content"][0]["text"]
assert message["role"] == "assistant"
- assert "" in text
- assert "result = search(query='GAIA')\nprint(result)" in text
+ assert "" not in text
+ assert "result = search(query=" in text and "GAIA" in text and "print(result)" in text
assert "python_interpreter'()" not in text
diff --git a/test/sdk/core/agents/test_context_manager_assembly.py b/test/sdk/core/agents/test_context_manager_assembly.py
index 1ae00ad618..f34ce49714 100644
--- a/test/sdk/core/agents/test_context_manager_assembly.py
+++ b/test/sdk/core/agents/test_context_manager_assembly.py
@@ -2,6 +2,9 @@
"""Focused tests for ContextManager-owned managed assembly."""
from __future__ import annotations
+import json
+from html import unescape
+
import pytest
from nexent.core.agents.context import (
ContextItemInput,
@@ -208,7 +211,6 @@ def test_context_manager_owns_final_answer_assembly():
"system",
"user",
"assistant",
- "assistant",
"user",
]
assert [_message_text(message) for message in final.messages[:3]] == [
@@ -216,7 +218,7 @@ def test_context_manager_owns_final_answer_assembly():
"final instruction",
"memory fact",
]
- action_history = _message_text(final.messages[4])
+ action_history = _message_text(final.messages[3])
assert '' in action_history
assert "work trace " in action_history
assert "Calling tools:" not in action_history
@@ -295,7 +297,8 @@ def test_completed_skill_read_is_available_to_the_next_model_step():
if '' in _message_text(message)
)
assert "read_skill_md" in str(final.messages)
- assert 'read_skill_md("demo-skill")' in '\n'.join(_message_text(m) for m in final.messages)
+ calls = json.loads(unescape(action_history.split("")[1].split(" ")[0]))
+ assert calls[0]["arguments"] == 'skill_content = read_skill_md("demo-skill")'
assert "SKILL_CONTENT_SENTINEL: follow the guide" in action_history
assert "Do not copy this record's format as your next response." not in action_history
diff --git a/test/sdk/core/agents/test_core_agent.py b/test/sdk/core/agents/test_core_agent.py
index 0e58faa3df..6a13f615ed 100644
--- a/test/sdk/core/agents/test_core_agent.py
+++ b/test/sdk/core/agents/test_core_agent.py
@@ -2569,8 +2569,8 @@ def test_step_stream_opens_turn_after_completed_assistant_action(self, monkeypat
assert "Do not repeat any completed action" in continuation
assert "" in continuation
- def test_step_stream_rolls_back_interstitial_text_before_protocol_repair(self):
- """Text between executable blocks is rolled back before protocol repair."""
+ def test_step_stream_rolls_back_malformed_code_before_protocol_repair(self):
+ """Invalid Python is rolled back before protocol repair."""
module = core_agent_module
agent = object.__new__(module.CoreAgent)
agent.enable_protocol_repair_retry = True
@@ -2592,7 +2592,7 @@ def test_step_stream_rolls_back_interstitial_text_before_protocol_repair(self):
agent.verification_controller = None
response = SimpleNamespace(
- content='print("first")explanationprint("ok")',
+ content='print("first")explanationprint("ok"',
token_usage=None,
model_attempt_id="semantic-attempt",
model_attempt_number=1,
diff --git a/test/sdk/core/agents/test_execution_history_compaction.py b/test/sdk/core/agents/test_execution_history_compaction.py
index e431998d79..cb4e9f1b84 100644
--- a/test/sdk/core/agents/test_execution_history_compaction.py
+++ b/test/sdk/core/agents/test_execution_history_compaction.py
@@ -27,7 +27,9 @@ def summary_model(**kwargs):
def item(fields, **kwargs):
return ContextItem.from_input(ContextItemInput(id='action', type='current_action', content={
- 'assistant_output': 'print(result)', 'execution_fields': fields, **kwargs,
+ 'assistant_output': 'print(result)',
+ 'tool_calls': [{'name': 'python_interpreter', 'arguments': 'print(result)'}],
+ 'execution_fields': fields, **kwargs,
}))
@@ -41,7 +43,7 @@ def test_ac002_values_are_not_dropped_or_given_guessed_types(value):
assert f'{value} ' in rendered
assert 'type=' not in rendered
assert rendered.count('print(result)') == 1
- assert 'completed_action_history' not in rendered
+ assert rendered.count('").removesuffix("")))
- for t in texts[:-1]] == calls
+ assert len(texts) == 1
+ assert json.loads(unescape(texts[0].split("")[1].split(" ")[0])) == calls
assert all("" not in t for t in texts)
assert "" in texts[-1] and "known error " in texts[-1]
assert item.content == original
@pytest.mark.parametrize("arguments", ["print('')", {"code": "print('')"}, ""])
-def test_ac_r5_only_interpreter_source_uses_code_envelope(arguments):
+def test_ac_r5_interpreter_source_is_read_only_data(arguments):
item = ContextItem.from_input(ContextItemInput(id="legacy", type="current_action", content={
"tool_calls": {"name": "python_interpreter", "arguments": arguments}, "result": "done",
}))
messages = ContextItemRenderer().render([item])
- source = arguments["code"] if isinstance(arguments, dict) else arguments
- assert messages[0]["content"][0]["text"] == f"\n{source}\n"
- assert messages[1]["content"][0]["text"].endswith("done \n ")
+ assert len(messages) == 1
+ text = messages[0]["content"][0]["text"]
+ assert "" not in text
+ assert json.loads(unescape(text.split("")[1].split(" ")[0])) == [
+ {"name": "python_interpreter", "arguments": arguments}
+ ]
+ assert "done " in text
+
def test_ac_r5_mixed_legacy_calls_keep_order_and_unknown_records_as_data():
@@ -70,11 +75,9 @@ def test_ac_r5_mixed_legacy_calls_keep_order_and_unknown_records_as_data():
}))
original = deepcopy(item.content)
texts = [m["content"][0]["text"] for m in ContextItemRenderer().render([item])]
- assert len(texts) == len(calls) + 1
- assert texts[1] == "\nprint(1)\n"
- for i in (0, 2, 3, 4, 5):
- assert texts[i].startswith("") and "" not in texts[i]
- assert json.loads(unescape(texts[i][13:-14])) == calls[i]
+ assert len(texts) == 1
+ assert "" not in texts[0]
+ assert json.loads(unescape(texts[0].split("")[1].split(" ")[0])) == calls
assert item.content == original
@@ -110,9 +113,13 @@ def test_ac_r5_authoritative_output_is_not_duplicated_by_fallback(passthrough):
content["messages"] = expected
item = ContextItem.from_input(ContextItemInput(id="legacy", type="current_action", content=content))
messages = ContextItemRenderer().render([item])
- assert messages[:1] == expected
- assert len(messages) == (1 if passthrough else 2)
- assert "" not in str(messages)
+ assert len(messages) == 1
+ if passthrough:
+ assert messages == expected
+ else:
+ assert "original" not in str(messages)
+ assert "" in str(messages)
+ assert "done " in str(messages)
@pytest.mark.parametrize("history_enabled", [False, True])
diff --git a/test/sdk/core/agents/test_guardrail_checkpoints.py b/test/sdk/core/agents/test_guardrail_checkpoints.py
index cc60c668ba..d6f4f862ca 100644
--- a/test/sdk/core/agents/test_guardrail_checkpoints.py
+++ b/test/sdk/core/agents/test_guardrail_checkpoints.py
@@ -377,7 +377,7 @@ def test_fault_injected_model_outputs_silently_repair_then_execute_explicit_fina
rule = GuardrailRule(name="irrelevant", pattern="never-match", severity="block")
agent = _make_step_agent(rule, messages=[_msg("user", "solve this")])
responses = []
- for content in ("42 "):
+ for content in ("{} ", "bare answer", "42 "):
response = MagicMock()
response.content = content
response.token_usage = None
diff --git a/test/sdk/core/agents/test_history_context_runtime.py b/test/sdk/core/agents/test_history_context_runtime.py
index 5759e80e86..d67c7b91fb 100644
--- a/test/sdk/core/agents/test_history_context_runtime.py
+++ b/test/sdk/core/agents/test_history_context_runtime.py
@@ -276,8 +276,8 @@ def test_projects_planning_and_multiple_actions_in_stable_run_order(monkeypatch)
assert "Calling tools:" not in rendered
assert "Observation:" not in rendered
- assert "reasoning 1" in rendered
- assert "reasoning 2" in rendered
+ assert "reasoning 1" not in rendered
+ assert "reasoning 2" not in rendered
assert result.evidence.item_representations == (
("current_task:0", "raw"), ("current_planning:0", "raw"),
("current_action:0", "raw"), ("current_action:1", "raw"),
@@ -317,15 +317,15 @@ def test_projects_tool_calls_as_json_serializable_payloads(monkeypatch):
model=None, memory=memory, current_run_start_idx=0, run_context=run,
)
rendered = _rendered_text(result.messages)
- assert "" in rendered
- assert "print('ok')" in rendered
+ assert "" not in rendered
+ assert "print(" in rendered and "ok" in rendered
assert "" in rendered
assert "done " in rendered
assert "Calling tools:" not in rendered
assert "Observation:" not in rendered
summary_rendered = _rendered_text(manager.render_memory_messages(memory))
- assert "" in summary_rendered
+ assert "" not in summary_rendered
assert "" in summary_rendered
assert "Calling tools:" not in summary_rendered
assert "Observation:" not in summary_rendered
@@ -422,7 +422,7 @@ def stage_observer(items, stable, dynamic, tools):
assert result.evidence.current_action_compact_count == 1
rendered = str(result.messages)
assert "old observation" in rendered
- assert "old reasoning" in rendered
+ assert "old reasoning" not in rendered
assert "knowledge knowledge" in rendered
diff --git a/test/sdk/core/agents/test_lowercase_output_protocol_isolated.py b/test/sdk/core/agents/test_lowercase_output_protocol_isolated.py
index fbba03ca6c..ea18c4c82d 100644
--- a/test/sdk/core/agents/test_lowercase_output_protocol_isolated.py
+++ b/test/sdk/core/agents/test_lowercase_output_protocol_isolated.py
@@ -1,3 +1,4 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
"""Exercise the output protocol without importing unrelated Nexent services."""
import importlib.util
@@ -40,6 +41,6 @@ def test_uppercase_final_answer_is_accepted(protocol):
assert answer.answer == "ok"
-def test_uppercase_legacy_run_is_rejected_even_after_code_label(protocol):
- with pytest.raises(protocol.ModelOutputProtocolError):
- protocol.classify_model_output("Code: ```print(1)```", protocol="code_action")
+def test_uppercase_legacy_run_is_accepted_after_code_label(protocol):
+ result = protocol.classify_model_output("Code: ```print(1)```", protocol="code_action")
+ assert result == protocol.ExecutableAction("print(1)", legacy_format=True)
diff --git a/test/sdk/core/agents/test_output_protocol.py b/test/sdk/core/agents/test_output_protocol.py
index 591d4c9fc1..2062b6a155 100644
--- a/test/sdk/core/agents/test_output_protocol.py
+++ b/test/sdk/core/agents/test_output_protocol.py
@@ -1,3 +1,4 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
"""Acceptance tests for the closed CodeAgent output protocol."""
import pytest
@@ -29,14 +30,7 @@ def test_ac_001_visible_content_is_meaningful(value):
("output", "reason"),
[
("\u200b\u2060\ufeff", ProtocolErrorReason.EMPTY_VISIBLE_CONTENT),
- ("unsupported ", ProtocolErrorReason.UNSUPPORTED_OR_TAG_ONLY_OUTPUT),
- ("print(1)", ProtocolErrorReason.MALFORMED_ACTION),
- ("print(1)suffix", ProtocolErrorReason.MALFORMED_ACTION),
- (
- "print(1)explanationprint(2)",
- ProtocolErrorReason.MALFORMED_ACTION,
- ),
("if True:print(1)",
@@ -44,7 +38,7 @@ def test_ac_001_visible_content_is_meaningful(value):
),
(
"done print('late')",
- ProtocolErrorReason.MALFORMED_ACTION,
+ ProtocolErrorReason.INVALID_FINAL_ENVELOPE,
),
(
"print(1)",
@@ -54,11 +48,7 @@ def test_ac_001_visible_content_is_meaningful(value):
"{} ",
ProtocolErrorReason.UNSUPPORTED_OR_TAG_ONLY_OUTPUT,
),
- ("<|python_tag|>{}", ProtocolErrorReason.UNSUPPORTED_OR_TAG_ONLY_OUTPUT),
- (
- "reason print(1)",
- ProtocolErrorReason.MALFORMED_ACTION,
- ),
+ ("<|python_tag|>{}", ProtocolErrorReason.MALFORMED_ACTION),
(
"<|channel|>analysisprint(1)",
ProtocolErrorReason.MALFORMED_ACTION,
@@ -115,11 +105,8 @@ def test_platform_think_code_preamble_is_not_an_invalid_action(prefix, legacy):
@pytest.mark.parametrize("output", [
- "Think: Done.\nCode:\nprint(1)\nObservation: fabricated",
- "Think: Example print(1)\nCode:\nprint(2)",
- "Think: Example ```python\nprint(1)\n```\nCode:\nprint(2)",
"Think: Need more data.\nCode:\nask_user(",
- "Think: Done.\nCode:\n```python\nprint(1)\n```",
+ "print(1)done suffix",
])
def test_preamble_does_not_relax_action_boundaries(output):
with pytest.raises(ModelOutputProtocolError):
@@ -235,7 +222,7 @@ def test_uppercase_final_envelope_is_accepted(protocol):
assert answer.answer == "ok"
-@pytest.mark.parametrize("protocol", ["final_envelope", "code_action"])
+@pytest.mark.parametrize("protocol", ["final_envelope"])
def test_cftp_001_lowercase_final_envelope_is_rejected(protocol):
with pytest.raises(ModelOutputProtocolError):
classify_model_output("legacy ", protocol=protocol)
diff --git a/test/sdk/core/agents/test_output_protocol_patch.py b/test/sdk/core/agents/test_output_protocol_patch.py
new file mode 100644
index 0000000000..95b0e8c0a4
--- /dev/null
+++ b/test/sdk/core/agents/test_output_protocol_patch.py
@@ -0,0 +1,128 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""v2.8.5 protocol boundaries and runtime-only history regressions."""
+
+import json
+from copy import deepcopy
+from html import unescape
+
+import pytest
+from smolagents.agents import EMPTY_PROMPT_TEMPLATES
+from smolagents.models import ChatMessage, Model
+from smolagents.tools import Tool
+
+from nexent.core.agents.context import ContextManager
+from nexent.core.agents.context.models import ContextItem, ContextItemInput
+from nexent.core.agents.context.rendering import ContextItemRenderer
+from nexent.core.agents.context.runtime import ManagedContextRuntime
+from nexent.core.agents.core_agent import CoreAgent
+from nexent.core.agents.output_protocol import (
+ ExecutableAction,
+ ModelOutputProtocolError,
+ NonterminalThought,
+ classify_model_output,
+)
+from nexent.core.utils.observer import MessageObserver
+
+
+@pytest.mark.parametrize("prefix", [
+ '思考：正文只有一个空的``标签。\n',
+ '思考：
data \n代码：',
+ 'inspect HTML \n',
+])
+@pytest.mark.parametrize("tags", [("", ""), ("", "")])
+def test_ac001_html_is_data_with_or_without_code_label(prefix, tags):
+ source = 'read_skill_md(skill_name="chapter-qa-mark-complete", additional_files=[])'
+ assert classify_model_output(prefix + tags[0] + source + tags[1], protocol="code_action") == ExecutableAction(source)
+
+
+@pytest.mark.parametrize("example", [
+ '`never()`',
+ '``never()``',
+ '```python\nnever()\n```',
+ '~~~xml\nnever()\n~~~',
+ '`fake `',
+])
+def test_ac003_markdown_examples_are_not_actions(example):
+ output = example + '\n代码：real()\nObservation: fabricated'
+ assert classify_model_output(output, protocol="code_action") == ExecutableAction('real()')
+ assert isinstance(classify_model_output(example, protocol="code_action"), NonterminalThought)
+
+
+def test_ac002_prose_between_actions_and_literal_tags_preserve_order():
+ output = 'value = "" Then use it.\nprint(value)Done.'
+ assert classify_model_output(output, protocol="code_action") == ExecutableAction('value = " "\n\nprint(value)')
+
+
+def test_ac002_backticks_inside_python_do_not_hide_later_actions():
+ output = 'print("`")print("`")'
+ assert classify_model_output(output, protocol="code_action").code == 'print("`")\n\nprint("`")'
+
+
+def test_ac004_combined_final_is_last_and_preserves_quoted_code():
+ output = 'run()Candidate:\nExample `x`. '
+ result = classify_model_output(output, protocol="code_action")
+ assert result == ExecutableAction('run()', final_answer='Example `x`.')
+
+
+@pytest.mark.parametrize("output", [
+ 'run()done trailing',
+ 'done run()',
+ 'run()run()one two ',
+ 'run(done ',
+ 'run()',
+ 'one(two()',
+])
+def test_ac004_invalid_structure_is_rejected_before_execution(output):
+ with pytest.raises(ModelOutputProtocolError):
+ classify_model_output(output, protocol="code_action")
+
+
+def test_ac005_only_actual_call_and_runtime_fields_enter_history():
+ item = ContextItem.from_input(ContextItemInput(id="step", type="current_action", content={
+ 'assistant_output': '工具成功 FAKE_RESULT actual()fake ',
+ 'tool_calls': [{'name': 'python_interpreter', 'arguments': 'actual()'}],
+ 'execution_fields': {'RETURN_VALUE': 'False', 'ERROR': 'actual failure'},
+ }))
+ messages = ContextItemRenderer().render([item])
+ assert len(messages) == 1 and messages[0]['role'] == 'assistant'
+ text = messages[0]['content'][0]['text']
+ assert 'FAKE_RESULT' not in text and '>fake<' not in text
+ assert text.count('') == 1
+ assert 'False ' in text and 'actual failure ' in text
+ assert json.loads(unescape(text.split('')[1].split(' ')[0]))[0]['arguments'] == 'actual()'
+
+
+@pytest.mark.parametrize('strict', [False, True])
+def test_ac004_mixed_candidate_executes_once_and_confirms_actual_result(strict):
+ calls = []
+ requests = []
+
+ class ActualTool(Tool):
+ name = 'actual'
+ description = 'Return the actual result.'
+ inputs = {}
+ output_type = 'string'
+
+ def forward(self):
+ calls.append('actual')
+ return 'REAL_FAILURE'
+
+ class RuntimeModel(Model):
+ def generate(self, messages, **kwargs):
+ requests.append(json.dumps(messages, default=str))
+ if len(requests) == 1:
+ return ChatMessage(role='assistant', content=(
+ 'print(actual())FAKE_SUCCESS '
+ ))
+ assert 'REAL_FAILURE' in requests[-1]
+ assert 'FAKE_SUCCESS' not in requests[-1]
+ return ChatMessage(role='assistant', content='Confirmed REAL_FAILURE ')
+
+ templates = deepcopy(EMPTY_PROMPT_TEMPLATES)
+ templates['system_prompt'] = 'Use runtime results.'
+ agent = CoreAgent(observer=MessageObserver(lang='en'), model=RuntimeModel(model_id='unit'),
+ tools=[ActualTool()], prompt_templates=templates, max_steps=3, verbosity_level=0,
+ context_runtime=ManagedContextRuntime(ContextManager(), []), enable_protocol_repair_retry=strict)
+ assert str(agent.run('Run actual once and report the result.')) == 'Confirmed REAL_FAILURE'
+ assert calls == ['actual'] and len(requests) == 2
diff --git a/test/sdk/core/agents/test_think_prefix_repair.py b/test/sdk/core/agents/test_think_prefix_repair.py
index 42a0b647e2..81fe1f67fa 100644
--- a/test/sdk/core/agents/test_think_prefix_repair.py
+++ b/test/sdk/core/agents/test_think_prefix_repair.py
@@ -1,3 +1,4 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
"""Unit regressions for the actual repair loop, with execution substituted."""
from types import SimpleNamespace
@@ -11,9 +12,9 @@
'read_skill_md(',
'Read the guide.read_skill_md(',
'print(1)explanationprint(2)',
- 'print(1)suffix',
- 'reason print(1)',
+ 'print(1)explanationprint(2',
+ 'print(1)done suffix',
+ 'reason print(1)',
'<|channel|>analysisprint(1)',
'partial',
'one two ',
diff --git a/test/sdk/core/utils/test_observer_mixed_answer.py b/test/sdk/core/utils/test_observer_mixed_answer.py
new file mode 100644
index 0000000000..40231ddac0
--- /dev/null
+++ b/test/sdk/core/utils/test_observer_mixed_answer.py
@@ -0,0 +1,35 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""Mixed action/final responses expose only post-execution confirmed answers."""
+import json
+
+import pytest
+
+from nexent.core.utils.observer import MessageObserver
+
+
+@pytest.mark.parametrize('chunk_size', [1, 3, 1000])
+@pytest.mark.parametrize('prefix', ['actual()', '```actual()```'])
+def test_mixed_answer_waits_for_runtime_result(prefix, chunk_size):
+ observer = MessageObserver(lang='en')
+ observer.begin_model_attempt('action', 1)
+ output = prefix + 'UNCONFIRMED '
+ for offset in range(0, len(output), chunk_size):
+ observer.add_model_new_token(output[offset:offset+chunk_size])
+ observer.flush_remaining_tokens()
+ events = [json.loads(value) for value in observer.message_query]
+ assert not any(event.get('type') == 'final_answer' for event in events)
+ observer.commit_model_attempt('action', 1)
+ observer.begin_model_attempt('confirmation', 1)
+ observer.add_model_new_token('CONFIRMED ')
+ observer.flush_remaining_tokens()
+ events = [json.loads(value) for value in observer.message_query]
+ assert ''.join(event.get('content', '') for event in events if event.get('type') == 'final_answer') == 'CONFIRMED'
+
+
+def test_quoted_action_example_does_not_suppress_normal_answer():
+ observer = MessageObserver(lang='en')
+ observer.begin_model_attempt('answer', 1)
+ observer.add_model_new_token('Example `display()`\nDISPLAY_ONLY ')
+ observer.flush_remaining_tokens()
+ events = [json.loads(value) for value in observer.message_query]
+ assert any(event.get('type') == 'final_answer' and event.get('content') == 'DISPLAY_ONLY' for event in events)
diff --git a/test/sdk/monitor/_http_response_lifecycle_cases.py b/test/sdk/monitor/_http_response_lifecycle_cases.py
new file mode 100644
index 0000000000..5ceb94a0ee
--- /dev/null
+++ b/test/sdk/monitor/_http_response_lifecycle_cases.py
@@ -0,0 +1,162 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""NB-AC01/03: real ASGI instrumentation covers the complete response send."""
+
+import asyncio
+
+import pytest
+from fastapi import FastAPI
+from fastapi.responses import JSONResponse, StreamingResponse
+from opentelemetry import trace
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+from opentelemetry.trace import SpanKind, StatusCode
+
+from nexent.monitor.monitoring import MonitoringConfig, get_monitoring_manager
+
+
+@pytest.fixture
+def instrument(monkeypatch):
+ exporter = InMemorySpanExporter()
+ provider = TracerProvider()
+ provider.add_span_processor(SimpleSpanProcessor(exporter))
+ monkeypatch.setattr(trace, 'get_tracer_provider', lambda: provider)
+ manager = get_monitoring_manager()
+ monkeypatch.setattr(manager, '_config', MonitoringConfig(enable_telemetry=True))
+ yield manager, exporter
+ provider.shutdown()
+
+
+def scope(path='/nb/v1/chat/run', headers=None):
+ return {'type': 'http', 'asgi': {'version': '3.0', 'spec_version': '2.4'},
+ 'http_version': '1.1', 'method': 'POST', 'scheme': 'http', 'path': path,
+ 'raw_path': path.encode(), 'query_string': b'', 'root_path': '',
+ 'headers': headers or [], 'server': ('northbound', 5013), 'client': ('client', 1234)}
+
+
+async def receive():
+ await asyncio.Event().wait()
+
+
+def server_spans(exporter):
+ return [span for span in exporter.get_finished_spans() if span.kind == SpanKind.SERVER]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('outcome', ['complete', 'send_error', 'cancel'])
+async def test_nb_ac001_span_stays_open_until_last_send(instrument, outcome):
+ manager, exporter = instrument
+ app = FastAPI()
+ last_send = asyncio.Event()
+ release = asyncio.Event()
+ chunks = []
+
+ @app.post('/nb/v1/chat/run')
+ async def run():
+ async def body():
+ yield b'data: first\n\n'
+ yield b'data: last\n\n'
+ return StreamingResponse(body(), media_type='text/event-stream')
+
+ assert manager.setup_fastapi_app(app)
+
+ async def send(message):
+ chunks.append(message)
+ if message['type'] == 'http.response.body' and not message.get('more_body', False):
+ last_send.set()
+ await release.wait()
+ if outcome == 'send_error':
+ raise RuntimeError('controlled transport failure')
+
+ task = asyncio.create_task(app(scope(), receive, send))
+ await asyncio.wait_for(last_send.wait(), 3)
+ assert not server_spans(exporter), 'Server span ended before last send completed'
+ assert b''.join(m.get('body', b'') for m in chunks) == b'data: first\n\ndata: last\n\n'
+ if outcome == 'cancel':
+ task.cancel()
+ with pytest.raises(asyncio.CancelledError):
+ await task
+ else:
+ release.set()
+ if outcome == 'send_error':
+ with pytest.raises(RuntimeError, match='controlled transport failure'):
+ await task
+ else:
+ await task
+ spans = server_spans(exporter)
+ assert len(spans) == 1 and spans[0].end_time >= spans[0].start_time
+ if outcome == 'send_error':
+ assert spans[0].status.status_code == StatusCode.ERROR
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('status', [200, 401, 503])
+async def test_nb_ac003_nonstream_and_http_errors_end_span(instrument, status):
+ manager, exporter = instrument
+ app = FastAPI()
+
+ @app.post('/nb/v1/chat/run')
+ async def run():
+ return JSONResponse({'result': 'unchanged'}, status_code=status)
+
+ assert manager.setup_fastapi_app(app)
+ messages = []
+
+ async def send(message):
+ messages.append(message)
+
+ await app(scope(), receive, send)
+ spans = server_spans(exporter)
+ assert len(spans) == 1
+ assert spans[0].attributes.get('http.status_code', spans[0].attributes.get('http.response.status_code')) == status
+ assert messages[0]['status'] == status
+ assert b'unchanged' in messages[-1]['body']
+
+
+@pytest.mark.asyncio
+async def test_nb_ac002_concurrent_requests_keep_separate_trace_context(instrument):
+ manager, exporter = instrument
+ app = FastAPI()
+ entered = 0
+ both = asyncio.Event()
+
+ @app.post('/nb/v1/chat/run')
+ async def run():
+ nonlocal entered
+ entered += 1
+ if entered == 2:
+ both.set()
+ await both.wait()
+ return JSONResponse({'ok': True})
+
+ assert manager.setup_fastapi_app(app)
+
+ async def send(_message):
+ pass
+
+ ids = ['1' * 32, '2' * 32]
+ await asyncio.gather(*(app(scope(headers=[(b'traceparent', f'00-{tid}-1111111111111111-01'.encode())]),
+ receive, send) for tid in ids))
+ spans = server_spans(exporter)
+ assert {f'{span.context.trace_id:032x}' for span in spans} == set(ids)
+ assert len({span.context.span_id for span in spans}) == 2
+
+
+@pytest.mark.asyncio
+async def test_nb_ac003_disabled_monitoring_keeps_response_without_spans(instrument, monkeypatch):
+ manager, exporter = instrument
+ monkeypatch.setattr(manager, '_config', MonitoringConfig(enable_telemetry=False))
+ app = FastAPI()
+
+ @app.post('/nb/v1/chat/run')
+ async def run():
+ return JSONResponse({'ok': True})
+
+ assert manager.setup_fastapi_app(app) is False
+ messages = []
+
+ async def send(message):
+ messages.append(message)
+
+ await app(scope(), receive, send)
+ assert messages[0]['status'] == 200 and not exporter.get_finished_spans()
diff --git a/test/sdk/monitor/test_http_response_lifecycle.py b/test/sdk/monitor/test_http_response_lifecycle.py
new file mode 100644
index 0000000000..c5741e160b
--- /dev/null
+++ b/test/sdk/monitor/test_http_response_lifecycle.py
@@ -0,0 +1,18 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""Run real OTel ASGI cases outside the global legacy OTel test stubs."""
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+
+def test_nb_ac001_real_asgi_response_lifecycle(tmp_path):
+ root = Path(__file__).resolve().parents[4]
+ cases = Path(__file__).with_name('_http_response_lifecycle_cases.py')
+ env = {**os.environ, 'PYTHONPATH': str(root/'sdk')}
+ result = subprocess.run([sys.executable, '-m', 'pytest', str(cases),
+ '--noconftest', '--basetemp', str(tmp_path/'child'), '-q'],
+ cwd=root, env=env, capture_output=True, text=True, timeout=60, check=False)
+ assert result.returncode == 0, result.stdout + result.stderr
+ assert '8 passed' in result.stdout