\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..d874f35fe8 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
@@ -39,6 +40,7 @@ class ExecutableAction:
code: str
legacy_format: bool = False
+ final_answer: str | None = None
@dataclass(frozen=True)
@@ -88,18 +90,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 +168,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 +229,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 +248,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)
diff --git a/sdk/nexent/core/models/openai_llm.py b/sdk/nexent/core/models/openai_llm.py
index 2e03925fd0..ad9e4db9af 100644
--- a/sdk/nexent/core/models/openai_llm.py
+++ b/sdk/nexent/core/models/openai_llm.py
@@ -501,10 +501,9 @@ def __call__(self, messages: List[Dict[str, Any]], stop_sequences: Optional[List
# Bound completion length unless the caller passed their own override
# via kwargs (which already landed in completion_kwargs above).
# OpenAI wire field stays max_tokens; internal name is max_output_tokens.
- # When a W2 snapshot is active, its requested_output_tokens is the sole
- # authority per CM-030 — skip the pre-W2 auto-fill so the dispatch
- # boundary does not see max_output_tokens masquerading as a caller
- # override and reject it via CallerMaxTokensOverrideForbidden.
+ # W2 decides whether a configured output cap should be dispatched.
+ # Skip pre-W2 auto-fill so a capacity ceiling cannot masquerade as an
+ # explicit caller cap or override an internal-only planning reserve.
if (
self.max_output_tokens is not None
and "max_tokens" not in completion_kwargs
@@ -789,11 +788,11 @@ def _close_stream_once():
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"requested_output_tokens": (
- self._coerce_context_budget_snapshot(
- trusted_budget_snapshot
- ).requested_output_tokens
- if trusted_budget_snapshot is not None
- else dispatch_kwargs.get("max_tokens")
+ self._provider_output_tokens(
+ self._coerce_context_budget_snapshot(trusted_budget_snapshot),
+ self.capacity_snapshot,
+ dispatch_kwargs.get("max_tokens"),
+ )
),
"reasoning_only_budget_exhausted": (
finish_reason == "length"
@@ -1113,9 +1112,9 @@ def _dispatch_chat_completion(
) -> Any:
"""Dispatch the OpenAI chat completion request.
- When W2 supplied a trusted safe-input-budget snapshot, this method is
- the provider dispatch boundary: caller `max_tokens` overrides must
- match the snapshot, and absent values are filled from the snapshot.
+ When W2 supplies a trusted context budget, explicit output caps must
+ match it. An implicit planning reserve remains internal when W1 has
+ no resolved default output reserve and the caller supplied no cap.
When the active W1 capacity snapshot is also threaded through, the
boundary additionally verifies W1->W2 fingerprint and provider/model
@@ -1145,7 +1144,13 @@ def _dispatch_chat_completion(
snapshot_value=trusted_max_tokens,
caller_value=caller_max_tokens,
)
- completion_kwargs["max_tokens"] = trusted_max_tokens
+ provider_max_tokens = self._provider_output_tokens(
+ snapshot, capacity_snapshot, caller_max_tokens
+ )
+ if provider_max_tokens is None:
+ completion_kwargs.pop("max_tokens", None)
+ else:
+ completion_kwargs["max_tokens"] = provider_max_tokens
logger.info(
"event=chat_completion_create model_id=%s kwargs=%s",
self.model_id,
@@ -1447,6 +1452,26 @@ def _sampling_fallback_kwargs(
if k not in ("temperature", "top_p")
}
+ @staticmethod
+ def _provider_output_tokens(
+ snapshot: Optional[ContextBudgetSnapshot],
+ capacity_snapshot: Optional[Dict[str, Any]],
+ caller_max_tokens: Optional[int],
+ ) -> Optional[int]:
+ """Keep an unconfigured planning reserve out of provider output caps."""
+ if snapshot is None:
+ return caller_max_tokens
+ if (
+ caller_max_tokens is None
+ and snapshot.output_reserve_source == "model_default"
+ and capacity_snapshot is not None
+ and "default_output_reserve_tokens" in capacity_snapshot
+ and capacity_snapshot["default_output_reserve_tokens"] is None
+ ):
+ return None
+ # Legacy snapshots without reserve metadata retain their cap semantics.
+ return snapshot.requested_output_tokens
+
@staticmethod
def _verify_w1_w2_consistency(
*,
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/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/AGENT-HISTORY-D3-001/case.yaml b/test-e2e/cases/AGENT-HISTORY-D3-001/case.yaml
index a357daa4e3..fce4aaf507 100644
--- a/test-e2e/cases/AGENT-HISTORY-D3-001/case.yaml
+++ b/test-e2e/cases/AGENT-HISTORY-D3-001/case.yaml
@@ -23,8 +23,9 @@ case:
- order: 1
action: Run short and multi-code-block tasks and inspect subsequent real model
input.
- expected: Code appears once, one CODE_EXEC_RESULT follows, false values and partial
- errors remain.
+ expected: One assistant read-only completed_action_history contains completed_action,
+ actual tool/input and outcome/recorded_result/recorded_error fields. Code and identical
+ result text appear once, including Docker OUTPUT versus its Python string representation in RETURN_VALUE; false values, distinct returns and partial errors remain.
- order: 2
action: Run a long result across the configured context threshold; inspect summary
call and repeated assembly.
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..2e1e9383d1
--- /dev/null
+++ b/test-e2e/cases/NB-OTEL-001/case.yaml
@@ -0,0 +1,88 @@
+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 and Runtime HTTP spans
+ each enclose normal Agent/model/tool work. Independent HTTP cleanup tails need
+ not end in parent-child order; actual model input and tokens are 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 and Runtime HTTP durations each enclose Agent/model/tool work
+ with shared trace and ancestry; independent HTTP cleanup end times are not ordered.
+ - 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..993f21524c
--- /dev/null
+++ b/test-e2e/cases/NB-OTEL-001/test.py
@@ -0,0 +1,161 @@
+# 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, *, require_runtime=False):
+ 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']
+ root_ready = any(expected_route in str(s['attributes'].get('http.route', '')) for s in spans)
+ runtime_ready = any('/agent/internal/northbound/run' in str(s['attributes'].get('http.route', '')) for s in spans)
+ if root_ready and (not require_runtime or runtime_ready):
+ 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', require_runtime=True)
+ 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']
+ # The peer can consume the final body before upstream ASGI cleanup returns.
+ assert root['start_time'] <= runtime[0]['start_time'] <= root['end_time']
+ # Sandbox warmup also carries model metadata but is not an LLM request.
+ models = [s for s in spans if s['attributes'].get('llm.operation.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']
+ assert runtime[0]['start_time'] <= span['start_time'] <= span['end_time'] <= runtime[0]['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/cases/TPC-D3-001/case.yaml b/test-e2e/cases/TPC-D3-001/case.yaml
index e61fb67bff..e62038f263 100644
--- a/test-e2e/cases/TPC-D3-001/case.yaml
+++ b/test-e2e/cases/TPC-D3-001/case.yaml
@@ -33,14 +33,18 @@ case:
a tool.
expected: No skill effect or completed/committed raw protocol final is emitted or stored. Provisional final deltas must
roll back in their exact attempts before the terminal error.
+ - order: 4
+ action: Return a legal skill action followed by model final-answer text in one response, then a separate response after the real result.
+ expected: Original model answer fragments stream normally before tool execution; the accepted skill still executes once, real evidence reaches the next model request and the ordinary loop completes.
expected_results:
- Runtime executes the skill read.
- History matches committed stream.
- Every rejected exhaustion is bounded to three generations and reports an error, with no skill effect or completed/committed
SSE final. Provisional final deltas are allowed only when their exact attempts explicitly roll back; history contains
only strictly empty compatibility final placeholders.
+ - Mixed-response streaming preserves model text without terminating the accepted action or hiding its real result.
forbidden_side_effects:
- - No unexecuted action final.
+ - No run terminates as completed before executing its accepted action.
- No duplicate tool effects.
integration_boundary:
- Provider to classifier
@@ -83,3 +87,13 @@ case:
forbidden_side_effects:
- 1
- 2
+
+ - id: AC-03
+ description: Mixed action/final text streams without the removed observer gate; actual skill execution and real-result continuation remain intact.
+ steps:
+ - 4
+ expected_results:
+ - 4
+ forbidden_side_effects:
+ - 1
+ - 2
diff --git a/test-e2e/cases/TPC-D3-001/execution.yaml b/test-e2e/cases/TPC-D3-001/execution.yaml
index eec22b7296..47f0b73870 100644
--- a/test-e2e/cases/TPC-D3-001/execution.yaml
+++ b/test-e2e/cases/TPC-D3-001/execution.yaml
@@ -17,6 +17,8 @@ preparation:
acceptance_bindings:
AC-01:
- test_tpc_d3_001_skill_executes_once_and_history_matches
+ AC-03:
+ - test_tpc_d3_001_mixed_answer_streams_and_action_continues
AC-02:
- test_tpc_d3_001_marked_exhaustion_never_executes_or_finalizes
timeout_seconds: 600
diff --git a/test-e2e/cases/TPC-D3-001/test.py b/test-e2e/cases/TPC-D3-001/test.py
index cb5cb49979..87d635e808 100644
--- a/test-e2e/cases/TPC-D3-001/test.py
+++ b/test-e2e/cases/TPC-D3-001/test.py
@@ -167,3 +167,37 @@ def test_tpc_d3_001_real_model_reads_skill(tmp_path, tpc_fixture):
request(cfg, 'DELETE', f'/conversation/{cid}')
pytestmark = [pytest.mark.case_id("TPC-D3-001"), pytest.mark.stage("D3")]
+
+
+@pytest.mark.parametrize('legacy', [False, True])
+def test_tpc_d3_001_mixed_answer_streams_and_action_continues(legacy, tmp_path, tpc_fixture):
+ cfg = tpc_fixture.copy()
+ code = CODE.replace('tpc-quality-check', cfg['skill_name'])
+ response_text = code + 'EARLY_MODEL_TEXT '
+ if legacy:
+ response_text = '```\n' + code.removeprefix('').removesuffix('') + '\n```EARLY_MODEL_TEXT '
+ request(cfg, 'POST', '/__control', {'responses': [response_text, '' + FINAL + ' '],
+ 'observation_marker': SKILL_MARKER}, provider=True)
+ cid = request(cfg, 'PUT', '/conversation/create', {'title': 'TPC ordinary mixed stream'})['data']['conversation_id']
+ try:
+ events = run_agent(cfg, cid)
+ history = request(cfg, 'GET', f'/conversation/{cid}')
+ stats = request(cfg, 'GET', '/__stats', provider=True)
+ evidence = Path(os.environ.get('NEXENT_TEST_ARTIFACT_DIR') or tmp_path)
+ evidence.mkdir(parents=True, exist_ok=True)
+ (evidence / f'tpc-mixed-{int(legacy)}.json').write_text(json.dumps(
+ {'events': events, 'history': history, 'provider': stats}, ensure_ascii=False, indent=2))
+ assert not any(e.get('type') == 'error' for e in events)
+ tool_indices = [i for i, e in enumerate(events) if e.get('type') == 'tool' and e.get('tool_name') == 'read_skill_md']
+ assert len(tool_indices) == 1
+ early = ''.join(str(e.get('content', '')) for e in events[:tool_indices[0]] if e.get('type') == 'final_answer')
+ assert early == 'EARLY_MODEL_TEXT'
+ assert stats['request_count'] == 2 and stats['observation_counts'] == [0, 1]
+ assert sum(e.get('type') == 'parse' for e in events) == 1
+ assert any(e.get('type') == 'execution_logs' and SKILL_MARKER in str(e.get('content')) for e in events)
+ for sequence in (events, units(history)):
+ assert final_answer_text(sequence) == FINAL
+ assert any(e.get('type') == 'execution_logs' and SKILL_MARKER in str(e.get('content')) for e in sequence)
+ finally:
+ request(cfg, 'DELETE', f'/conversation/{cid}')
+ request(cfg, 'POST', '/__control', {'responses': ['TPC ready ']}, provider=True)
diff --git a/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/case.yaml b/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/case.yaml
index ad4255e0de..3930f56a07 100644
--- a/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/case.yaml
+++ b/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/case.yaml
@@ -6,6 +6,7 @@ case:
feature_id: F-171
business_rule_ids:
- F-171-RULE-01
+ - F-171-RULE-02
title: W2 预算契约字段改名：context_budget_snapshot 序列化、parse 校验与派发边界｜核心流程
type: SDK-UT
priority: P1
@@ -47,7 +48,7 @@ case:
action: 派发边界 max_tokens：构造 snapshot(requested_output_tokens=X)，用 mock 的 self.client.chat.completions.create 调用 _dispatch_chat_completion，分别验证调用方
max_tokens 与 X 不一致时抛 CallerMaxTokensOverrideForbidden(snapshot_value=X, caller_value=传入值)、未传 max_tokens 时 completion_kwargs
被填充为 X。
- expected: 未传 max_tokens 时 dispatch 以 snapshot.requested_output_tokens 作为唯一权威填充 completion。
+ expected: 显式预算或缺少旧版 W1 预留元数据时按快照填充；W1 明确无默认预留且无显式预算时省略 max_tokens，保留内部上下文预算。
- order: 7
action: 派发边界 W1-W2 一致性：传入 capacity_snapshot（capacity_fingerprint/provider/model_name）与快照 w1_fingerprint/provider/model_name
不一致时，断言抛 ContextBudgetCapacityMismatch 且 field 为 w1_fingerprint/provider/model_name 中的对应项；capacity_snapshot 为空或三字段均缺省时跳过该校验（迁移窗口）。
@@ -58,6 +59,9 @@ case:
等新键；全程无未处理异常、无明文 secret 输出；测试结束释放 mock 对象，确保可重复执行。
expected: 追踪属性使用新字段名（w2.budget_fingerprint/w2.w1_fingerprint/context.effective_input_limit_tokens 等）；日志/Trace 不出现明文 API
Key/Token/Password/模型密钥。
+ - order: 9
+ action: 以真实容量解析器、预算计算器、AgentConfig 和模型适配器构建空预留及显式预留变体，通过受控 HTTP transport 检查 OpenAI 客户端实际 JSON；核对输入上限、压缩阈值及快照往返。
+ expected: 空预留不发送隐式 max_tokens；显式预留正常发送；内部预算、指纹及上下文阈值保留。
expected_results:
- AgentConfig/AgentRunInfo 的序列化键为 context_budget_snapshot，旧键 safe_input_budget_snapshot 不出现；process-local 字段不落入序列化输出。
- parse_context_budget_snapshot 对一致性字典返回 ContextBudgetSnapshot，新字段名 effective_input_limit_tokens/compaction_trigger_threshold_tokens/compaction_target_tokens
@@ -66,8 +70,9 @@ case:
- OpenAIModel.context_budget_snapshot 属性每次运行时赋值都经 parse 校验并回读为 ContextBudgetSnapshot。
- 派发边界：调用方 max_tokens 与 snapshot.requested_output_tokens 冲突时抛 CallerMaxTokensOverrideForbidden；W1-W2 指纹/provider/model 不一致时抛
ContextBudgetCapacityMismatch；无 capacity_snapshot 时跳过一致性校验。
- - 未传 max_tokens 时 dispatch 以 snapshot.requested_output_tokens 作为唯一权威填充 completion。
+ - 显式预算或缺少旧版 W1 预留元数据时按快照填充；W1 明确无默认预留且无显式预算时省略 max_tokens，保留内部上下文预算。
- 追踪属性使用新字段名（w2.budget_fingerprint/w2.w1_fingerprint/context.effective_input_limit_tokens 等）；日志/Trace 不出现明文 API Key/Token/Password/模型密钥。
+ - 空预留不发送隐式 max_tokens；显式预留正常发送；内部预算、指纹及上下文阈值保留。
forbidden_side_effects: []
source:
legacy_workbook_sha256: aa4b32414d30f12b3b64e14236268145e556bd7cfc5d3eb8df80731608ebc022
@@ -84,5 +89,33 @@ case:
- OpenAIModel.context_budget_snapshot 属性每次运行时赋值都经 parse 校验并回读为 ContextBudgetSnapshot。
- 派发边界：调用方 max_tokens 与 snapshot.requested_output_tokens 冲突时抛 CallerMaxTokensOverrideForbidden；W1-W2 指纹/provider/model 不一致时抛
ContextBudgetCapacityMismatch；无 capacity_snapshot 时跳过一致性校验。
- - 未传 max_tokens 时 dispatch 以 snapshot.requested_output_tokens 作为唯一权威填充 completion。
+ - 显式预算或缺少旧版 W1 预留元数据时按快照填充；W1 明确无默认预留且无显式预算时省略 max_tokens，保留内部上下文预算。
- 追踪属性使用新字段名（w2.budget_fingerprint/w2.w1_fingerprint/context.effective_input_limit_tokens 等）；日志/Trace 不出现明文 API Key/Token/Password/模型密钥。
+ acceptance:
+ - id: AC-01
+ description: Preserve existing serialization, fingerprint and dispatch authority contracts.
+ steps:
+ - 1
+ - 2
+ - 3
+ - 4
+ - 5
+ - 6
+ - 7
+ - 8
+ expected_results:
+ - 1
+ - 2
+ - 3
+ - 4
+ - 5
+ - 6
+ - 7
+ forbidden_side_effects: []
+ - id: AC-02
+ description: Omit implicit wire caps without removing internal context accounting.
+ steps:
+ - 9
+ expected_results:
+ - 8
+ forbidden_side_effects: []
diff --git a/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/execution.yaml b/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/execution.yaml
index b9c71200d1..307dcb63b4 100644
--- a/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/execution.yaml
+++ b/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/execution.yaml
@@ -5,3 +5,8 @@ implementations:
file: test.py
selector: UT-SDK-AUTO-74430BEE8438F486
notes: V5 source binding migrated; runtime parity is not yet certified.
+acceptance_bindings:
+ AC-01:
+ - test_w2_context_budget_snapshot_rename_and_dispatch_boundary
+ AC-02:
+ - test_implicit_output_budget_preserves_context_and_wire_contract
diff --git a/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/test.py b/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/test.py
index 3c4fa8bc0d..d2f4893d53 100644
--- a/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/test.py
+++ b/test-e2e/cases/UT-SDK-AUTO-74430BEE8438F486/test.py
@@ -4,8 +4,8 @@
Covers the V2 ``context_budget_snapshot`` field rename from the legacy
``safe_input_budget_snapshot`` across serialization, parse validation, and the
-trusted provider dispatch boundary. Pure in-memory tests; no network, no real
-model, no production credentials.
+trusted provider dispatch boundary. In-memory and controlled HTTP-transport tests; no external network, real
+model or production credentials.
"""
from __future__ import annotations
@@ -265,3 +265,63 @@ def test_w2_context_budget_snapshot_rename_and_dispatch_boundary():
assert attrs["context.compaction_target_tokens"] == trace_snapshot["compaction_target_tokens"]
assert attrs["w2.requested_output_tokens"] == trace_snapshot["requested_output_tokens"]
assert OpenAIModel._context_budget_trace_attributes(None) == {}
+
+
+@pytest.mark.case_id("UT-SDK-AUTO-74430BEE8438F486")
+@pytest.mark.stage("D1")
+@pytest.mark.parametrize("default_reserve", [None, 4096, 8192])
+def test_implicit_output_budget_preserves_context_and_wire_contract(default_reserve):
+ """Compose real capacity, Agent and dispatch contracts at a controlled HTTP boundary."""
+ import json
+
+ import httpx
+ from openai import OpenAI
+
+ from nexent.core.models.capacity_budget import CapacityReservePolicy, ContextBudgetCalculator
+ from nexent.core.models.capacity_resolver import resolve_capacity
+
+ overrides = {"context_window_tokens": 32768, "max_output_tokens": 16384}
+ if default_reserve is not None:
+ overrides["default_output_reserve_tokens"] = default_reserve
+ capacity = resolve_capacity(model_id="gpt-4o-mini", provider="openai", capability_profiles={},
+ operator_overrides=overrides)
+ budget = ContextBudgetCalculator().calculate_context_budget(
+ capacity_snapshot=capacity, reserve_policy=CapacityReservePolicy()
+ )
+ agent = _make_agent_config(budget.model_dump())
+ parsed = parse_context_budget_snapshot(agent.model_dump()["context_budget_snapshot"])
+ reserve = 4096 if default_reserve is None else default_reserve
+ assert parsed.requested_output_tokens == reserve
+ assert parsed.effective_input_limit_tokens == 32768 - reserve
+ assert parsed.compaction_trigger_threshold_tokens == int((32768 - reserve) * 0.8)
+ assert parsed.compaction_target_tokens == int((32768 - reserve) * 0.6)
+ assert parsed.uncertainty_reserve_tokens > 0
+ assert parsed.fingerprint == budget.fingerprint
+ captured = []
+
+ def provider(request):
+ captured.append(json.loads(request.content))
+ return httpx.Response(200, json={
+ "id": "fixture-completion", "object": "chat.completion", "created": 0,
+ "model": "gpt-4o-mini", "choices": [
+ {"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}
+ ],
+ })
+
+ model = _bare_model()
+ model.context_budget_snapshot = parsed
+ projection = {"capacity_fingerprint": capacity.fingerprint, "provider": capacity.provider,
+ "model_name": capacity.model_name, "default_output_reserve_tokens": default_reserve}
+ with OpenAI(api_key="fixture-only", base_url="http://provider.invalid/v1",
+ http_client=httpx.Client(transport=httpx.MockTransport(provider))) as client:
+ model.client = client
+ response = model._dispatch_chat_completion(
+ context_budget_snapshot=model.context_budget_snapshot, capacity_snapshot=projection,
+ model="gpt-4o-mini", messages=[{"role": "user", "content": "hello"}],
+ )
+ assert response.choices[0].message.content == "ok" and len(captured) == 1
+ if default_reserve is None:
+ assert "max_tokens" not in captured[0]
+ else:
+ assert captured[0]["max_tokens"] == default_reserve
+ assert model.context_budget_snapshot == parsed
diff --git a/test-e2e/changes/bugs/HISTORY-RENDERING-20261112.yaml b/test-e2e/changes/bugs/HISTORY-RENDERING-20261112.yaml
new file mode 100644
index 0000000000..c73a03624a
--- /dev/null
+++ b/test-e2e/changes/bugs/HISTORY-RENDERING-20261112.yaml
@@ -0,0 +1,43 @@
+schema_version: '1.0'
+change:
+ change_id: HISTORY-RENDERING-20261112
+ change_type: bugfix
+ bug_id: HISTORY-RENDERING-20261112
+ title: Restore v2.8.2 action-history presentation and deduplicate result text
+ description: Restore read-only completed_action fields while retaining canonical results and all existing summary/recall capabilities.
+ observed_behavior: Current history uses TOOL_RECORD and CODE_EXEC_RESULT rather than the requested v2.8.2 presentation.
+ expected_behavior: One assistant history record preserves actual tool/input/output/error, removes only equal return/output duplication and retains summary/recall.
+ escaped_stage: D3
+ coverage_gap:
+ type: missing_assertion
+ description: Existing regressions enforce the replaced presentation and lack exact-equality rendering deduplication assertions, including the Docker Jupyter string representation.
+ feature_contract_changed: false
+ affected_features:
+ existing: [F-171]
+ added: []
+ modified: []
+ retired: []
+ affected_cases:
+ existing: []
+ added: []
+ modified: [AGENT-HISTORY-D3-001]
+ retired: []
+ traditional_ut:
+ decision: strengthen
+ behaviors:
+ - Restore legacy read-only action fields and actual step identity without replaying model narration.
+ - Deduplicate equal stdout/return text while preserving different values, repeated logs, errors and masked output.
+ - Verify actual execution count and next-model-request results in strict and legacy modes.
+ - Preserve summaries, previews and authorized original-result recall.
+ test_selectors:
+ - test/sdk/core/agents/test_history_rendering_v282.py
+ - test/sdk/core/agents/test_output_protocol_patch.py
+ - test/sdk/core/agents/test_execution_history_compaction.py
+ - test/sdk/core/agents/test_execution_history_review_regressions.py
+ reason: Renderer unit behavior and existing real internal Agent collaboration already have owners; no duplicate formal D1.
+ notes: >-
+ SPEC nexent-doc/nexent/codeagent-action-history-role/changes/chat-action-history-rendering/design.md.
+ User explicitly preserves parsing/execution/summary/recall and Northbound/OTel.
+ Docker runtime acceptance additionally exposed a canonical OUTPUT/string-repr RETURN_VALUE duplicate; strengthen renderer-only equality and unit coverage without changing original execution records.
+ Update only presentation expectations in existing manual AGENT-HISTORY-D3-001; preserve other obligations.
+ UT results and manual/model-runtime execution evidence must be reported separately.
diff --git a/test-e2e/changes/bugs/IMPLICIT-OUTPUT-CAP-20261113.yaml b/test-e2e/changes/bugs/IMPLICIT-OUTPUT-CAP-20261113.yaml
new file mode 100644
index 0000000000..63e592cd19
--- /dev/null
+++ b/test-e2e/changes/bugs/IMPLICIT-OUTPUT-CAP-20261113.yaml
@@ -0,0 +1,61 @@
+schema_version: '1.0'
+change:
+ change_id: IMPLICIT-OUTPUT-CAP-20261113
+ change_type: bugfix
+ bug_id: IMPLICIT-OUTPUT-CAP-20261113
+ title: Do not send an implicit completion cap for an absent output reserve
+ description: Separate the implicit internal context planning reserve from an explicitly
+ configured provider output cap.
+ observed_behavior: A missing default_output_reserve_tokens becomes max_tokens=4096
+ at dispatch and can exhaust reasoning before visible output.
+ expected_behavior: Omit the implicit wire cap when the resolved default reserve
+ is absent, preserving internal context accounting and explicit budgets.
+ escaped_stage: D1
+ coverage_gap:
+ type: missing_assertion
+ description: Existing dispatch tests require snapshot caps but lack a null resolved
+ reserve end-to-end component variant.
+ feature_contract_changed: true
+ affected_features:
+ existing:
+ - F-183
+ added: []
+ modified:
+ - F-171
+ retired: []
+ affected_cases:
+ existing:
+ - UT-SDK-AUTO-5DD3BD24C1EF4718
+ - UT-SDK-AUTO-9B9B36B05DA14338
+ - UT-SDK-AUTO-FD74C47E39BAE2FC
+ - UT-SDK-AUTO-2EBD3B18F61B382A
+ - UT-SDK-AUTO-C80B9042CCD2DE71
+ - UT-SDK-AUTO-85A0FBE8116B9F9A
+ - UT-SDK-AUTO-0704DCE86C3E5C77
+ added: []
+ modified:
+ - UT-SDK-AUTO-74430BEE8438F486
+ retired: []
+ traditional_ut:
+ decision: strengthen
+ behaviors:
+ - Absent reserves omit only implicit wire caps
+ - Explicit budgets and legacy calls retain existing caps
+ - Fingerprint and caller-override rejection precede dispatch
+ - Empty reserve preserves context budgeting, compression, final-fit and overflow
+ recovery
+ test_selectors:
+ - test/sdk/core/models/test_implicit_output_cap.py
+ - test/sdk/core/models/test_openai_llm.py
+ - test/sdk/core/models/test_capacity_budget.py
+ - test/sdk/core/models/test_capacity_resolver.py
+ - test/sdk/core/agents/test_history_context_runtime.py
+ reason: Focused UT prove branch/error boundaries; the existing D1 adds actual
+ resolver-to-budget-to-adapter serialization evidence with a controlled HTTP
+ boundary.
+ notes: SPEC ../nexent-doc/docs/Developing/context-budget/request-budget-recovery/changes/implicit-output-cap/design.md.
+ No new D1; strengthen the existing component case and preserve all migrated Cases.
+ Reuse budget, fingerprint, compaction, overflow and migration D1 as adjacent context
+ regressions. No product HTTP/browser/deployment change, so D2-D5 are not selected.
+ Internal planning reserve is not a provider output guarantee when max_tokens is
+ omitted.
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..2f595b1f5b
--- /dev/null
+++ b/test-e2e/changes/bugs/NB-OTEL-20261111.yaml
@@ -0,0 +1,54 @@
+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.
+ Independent Northbound tracing change split from PR4328; no tool-history or output-parser changes. 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.
+ Observer-withdrawal regression rerun exposed an evidence polling race: the Northbound root batch can be ingested before the Runtime batch. Normal-stream trace polling now waits for the Runtime HTTP span as well, preserving ancestry assertions and the unauthorized/disconnect readiness contract. Cross-process ASGI cleanup can finish just after the peer consumes the response. Assert that both HTTP spans enclose all actual Agent/model/tool work instead of requiring ordered cleanup tails. Sandbox warmup can carry llm.model_name without being an LLM request; select actual requests by llm.operation.name before checking input and token evidence. No OTel product code changed.'
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..d297a441e3
--- /dev/null
+++ b/test-e2e/changes/bugs/OUTPUT-PROTOCOL-HISTORY-20261111.yaml
@@ -0,0 +1,55 @@
+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-D4-001
+ - AGENT-HISTORY-D3-001
+ added: []
+ modified:
+ - TPC-D3-001
+ retired: []
+ traditional_ut:
+ decision: strengthen
+ behaviors:
+ - Top-level case-insensitive actions with prose and HTML
+ - Markdown examples never execute
+ - Mixed model answer text streams normally; code executes once and real results reach ordinary continuation
+ - 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/utils/test_observer_mixed_answer.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.
+ User clarified on 2026-11-13 that prompt confirmation guidance stays, but the new observer stream suppression is withdrawn. Restore baseline observer behavior without changing parsing, tool execution, canonical results or the ordinary CodeAgent loop. The structured-output configuration regression remains an explicitly accepted known limitation and is not repaired in this change. Strengthen existing TPC-D3-001 for mixed streaming and continuation; reuse existing UT and add no duplicate D1.
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..c1f408d2bd 100644
--- a/test-e2e/features/F-146/feature.yaml
+++ b/test-e2e/features/F-146/feature.yaml
@@ -12,6 +12,10 @@ feature:
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.
+ - 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/F-171/feature.yaml b/test-e2e/features/F-171/feature.yaml
index db7e4ef5d6..2743384388 100644
--- a/test-e2e/features/F-171/feature.yaml
+++ b/test-e2e/features/F-171/feature.yaml
@@ -10,6 +10,11 @@ feature:
business_rules:
- rule_id: F-171-RULE-01
description: agent/tool/context gateway 契约：上下文预算与压缩、受管 MCP/模型并发、沙箱桥接线程回收、工具与观察者消息
+ - rule_id: F-171-RULE-02
+ description: When the resolved default output reserve is absent and no explicit
+ output budget is supplied, omit the implicit provider max_tokens cap while preserving
+ internal context accounting, final-fit, compaction and bounded overflow recovery.
+ Explicit budgets remain authoritative.
required_test_stages:
- D1
source:
diff --git a/test-e2e/features/TPC-001/feature.yaml b/test-e2e/features/TPC-001/feature.yaml
index e01b1981f7..6fd67c0f90 100644
--- a/test-e2e/features/TPC-001/feature.yaml
+++ b/test-e2e/features/TPC-001/feature.yaml
@@ -10,14 +10,17 @@ 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. Prompt guidance requests actual-result
+ confirmation, while runtime follows the ordinary CodeAgent loop.
- rule_id: TPC-RULE-002
description: Failed protocol-marked output cannot become a raw final at exhaustion;
valid explicit completion retains its envelope boundary.
- rule_id: TPC-RULE-003
- description: Tool effects occur once and committed stream/history reflects actions
- followed by explicit completion.
+ description: Tool effects occur once and real results enter history before ordinary
+ CodeAgent continuation. Model final-answer streaming is not suppressed solely
+ because executable code precedes it.
required_test_stages:
- D3
- D4
diff --git a/test-e2e/infra/automation/d3/think_prefix_mock_server.py b/test-e2e/infra/automation/d3/think_prefix_mock_server.py
index 302e63c24a..a20c44fece 100644
--- a/test-e2e/infra/automation/d3/think_prefix_mock_server.py
+++ b/test-e2e/infra/automation/d3/think_prefix_mock_server.py
@@ -1,3 +1,4 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
"""TPC controlled provider: exact malformed/valid content, real deployed executor.
Run: python test-e2e/infra/automation/d3/think_prefix_mock_server.py --port 19091
@@ -16,6 +17,12 @@
class Handler(OpenAICompatibleMockHandler):
+ def do_GET(self):
+ if urlsplit(self.path).path == '/__stats' and getattr(self.server, 'observation_marker', None):
+ self._send_json(200, {**self.state.snapshot(), 'observation_counts': self.server.observation_counts})
+ else:
+ super().do_GET()
+
def do_POST(self):
payload = self._read_json()
path = urlsplit(self.path).path
@@ -24,6 +31,12 @@ def do_POST(self):
if not responses or not all(isinstance(item, str) for item in responses):
self._send_error(400, 'invalid_request_error', 'responses must be nonempty strings')
return
+ marker = payload.get('observation_marker')
+ if marker is not None and (not isinstance(marker, str) or not 1 <= len(marker) <= 256):
+ self._send_error(400, 'invalid_request_error', 'Invalid observation marker')
+ return
+ self.server.observation_marker = marker
+ self.server.observation_counts = []
self.state.configure({})
self.server.responses = responses
self.server.finish_reason = payload.get('finish_reason', 'stop')
@@ -40,6 +53,8 @@ def do_POST(self):
finish = 'stop'
else:
number, _ = self.state.record(payload, bool(self.headers.get('Authorization')))
+ if getattr(self.server, 'observation_marker', None):
+ self.server.observation_counts.append(json.dumps(payload.get('messages', []), ensure_ascii=False).count(self.server.observation_marker))
text = self.server.responses[min(number - 1, len(self.server.responses) - 1)]
finish = self.server.finish_reason
if payload.get('stream'):
diff --git a/test-e2e/infra/environment/services/test-assets/tpc_provider.py b/test-e2e/infra/environment/services/test-assets/tpc_provider.py
index 3553754a8e..b330f1e280 100644
--- a/test-e2e/infra/environment/services/test-assets/tpc_provider.py
+++ b/test-e2e/infra/environment/services/test-assets/tpc_provider.py
@@ -22,7 +22,10 @@ async def configure(nonce: str, request: Request):
raise HTTPException(400, 'Invalid sequence')
if not all(isinstance(item, str) for item in responses):
raise HTTPException(400, 'Responses must be strings')
- scenarios[nonce] = {'responses': responses, 'request_count': 0,
+ marker = body.get('observation_marker')
+ if marker is not None and (not isinstance(marker, str) or not 1 <= len(marker) <= 256):
+ raise HTTPException(400, 'Invalid observation marker')
+ scenarios[nonce] = {'observation_marker': marker, 'observation_counts': [], 'responses': responses, 'request_count': 0,
'finish_reason': body.get('finish_reason', 'stop'),
'auxiliary_title': body.get('auxiliary_title', 'TPC verification')}
return {'ready': True}
@@ -33,7 +36,9 @@ async def stats(nonce: str, request: Request):
controlled(request)
if nonce not in scenarios:
raise HTTPException(404, 'Unknown sequence')
- return {'request_count': scenarios[nonce]['request_count']}
+ state = scenarios[nonce]
+ observed = {'observation_counts': state['observation_counts']} if state['observation_marker'] else {}
+ return {'request_count': state['request_count'], **observed}
@router.delete('/tpc/{nonce}')
@@ -52,6 +57,8 @@ async def complete(nonce: str, request: Request):
if body.get('max_tokens') == 4096:
index = state['request_count']
state['request_count'] += 1
+ if state['observation_marker']:
+ state['observation_counts'].append(json.dumps(body.get('messages', []), ensure_ascii=False).count(state['observation_marker']))
text = state['responses'][min(index, len(state['responses']) - 1)]
finish = state['finish_reason']
else:
diff --git a/test-e2e/infra/tools/tests/test_oct22_agent_triage.py b/test-e2e/infra/tools/tests/test_oct22_agent_triage.py
index 502860a63d..d078c15835 100644
--- a/test-e2e/infra/tools/tests/test_oct22_agent_triage.py
+++ b/test-e2e/infra/tools/tests/test_oct22_agent_triage.py
@@ -155,6 +155,11 @@ async def test_tpc_provider_keeps_auxiliary_calls_out_of_sequence_and_preserves_
second = await api.post(path + '/v1/chat/completions', json={'max_tokens': 4096, 'model': 'tpc'})
assert second.json()['choices'][0]['message']['content'] == 'OK '
assert (await api.get(path + '/__stats')).json()['request_count'] == 2
+ assert (await api.post(path + '/__control', json={'responses': ['OK '], 'observation_marker': 123})).status_code == 400
+ assert (await api.post(path + '/__control', json={'responses': ['OK '], 'observation_marker': 'REAL_MARKER'})).status_code == 200
+ await api.post(path + '/v1/chat/completions', json={'max_tokens': 4096, 'model': 'tpc', 'messages': []})
+ await api.post(path + '/v1/chat/completions', json={'max_tokens': 4096, 'model': 'tpc', 'messages': [{'role': 'assistant', 'content': 'REAL_MARKER'}]})
+ assert (await api.get(path + '/__stats')).json()['observation_counts'] == [0, 1]
assert (await api.delete(path)).status_code == 200
assert (await api.get(path + '/__stats')).status_code == 404
remote = httpx.ASGITransport(app=app, client=('192.0.2.1', 1234))
@@ -206,3 +211,31 @@ async def get(path):
async with runtime_fault.owned_failure_model(owner, 'AGT-015'):
pytest.fail('Unadmitted fixture reached inference')
assert writes[-1][0] == '/model/delete'
+
+
+def test_northbound_trace_poll_waits_for_runtime_batch(monkeypatch, tmp_path):
+ import importlib.util
+ case = Path(__file__).resolve().parents[3] / 'cases/NB-OTEL-001/test.py'
+ spec = importlib.util.spec_from_file_location('nb_trace_polling', case)
+ module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(module)
+ trace_id = 'f' * 32
+ def span(route, span_id):
+ return {'name': route, 'attributes': {'http.route': route},
+ 'context': {'trace_id': trace_id, 'span_id': span_id},
+ 'start_time': '2026-11-13T00:00:00Z', 'end_time': '2026-11-13T00:00:01Z'}
+ north = span('/nb/v1/chat/run', '1' * 16)
+ runtime = span('/agent/internal/northbound/run', '2' * 16)
+ batches = iter([[north], [north, runtime]])
+ calls = []
+ def get(*args, **kwargs):
+ calls.append('read')
+ data = next(batches)
+ return SimpleNamespace(status_code=200, json=lambda: {'data': data})
+ monkeypatch.setattr(module.requests, 'get', get)
+ monkeypatch.setattr(module.time, 'sleep', lambda *_: None)
+ monkeypatch.setenv('NEXENT_TEST_ARTIFACT_DIR', str(tmp_path))
+ result = module.read_trace({'phoenix_url': 'http://fixture', 'phoenix_project': 'test',
+ 'phoenix_api_key': 'fixture-phoenix', 'northbound_api_key': 'fixture-northbound'},
+ trace_id, '/nb/v1/chat/run', require_runtime=True)
+ assert result == [north, runtime] and calls == ['read', 'read']
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..c7874afef7
--- /dev/null
+++ b/test/backend/services/test_runtime_proxy_telemetry.py
@@ -0,0 +1,86 @@
+# 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/nb/v1/generate_title')
+ assert not re.search(excluded, 'http://northbound:5013/api/nb/v1/generate_title')
+ 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')
+ assert re.search(excluded, 'http://northbound:5013/api/health')
+ for route in ('/nb/v1/chat/run', '/nb/v1/generate_title'):
+ explicit = monitoring._build_fastapi_excluded_urls(included, route)
+ assert any(re.search(pattern, 'http://northbound:5013/api' + route) 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..5346a5b3b5 100644
--- a/test/sdk/core/agents/test_context_helper_contracts.py
+++ b/test/sdk/core/agents/test_context_helper_contracts.py
@@ -309,9 +309,9 @@ def test_renderer_current_action_without_raw_messages():
message = ContextItemRenderer().render([action])[0]
assert message["role"] == "assistant"
text = message["content"][0]["text"]
- assert '' in text
- assert "index: 1" not in text
- assert "done " in text
+ assert '' in text
+ assert "index: 1" in text
+ assert "recorded_result:\ndone" in text
assert "Step 1:" not in text
assert not text.lstrip().startswith("{")
@@ -341,11 +341,11 @@ def test_renderer_current_action_compact_with_tool_calls():
)
message = ContextItemRenderer().render([action])[0]
text = "\n".join(m["content"][0]["text"] for m in ContextItemRenderer().render([action]))
- assert "index: 3" not in text
- assert "" in text and "" not in text
- assert '"name": "search"' in unescape(text) and '"q": "foo"' in unescape(text)
- assert "" in text
- assert "answer is 42 " in text
+ assert "index: 3" in text
+ assert "input:\n" in text and "" not in text
+ assert 'tool: search' in text and '"q": "foo"' in unescape(text)
+ assert "outcome:\nfound 5 results" in text
+ assert "recorded_result:\nanswer is 42" in text
assert "Called tool" not in text
assert not text.lstrip().startswith("{")
@@ -362,8 +362,8 @@ def test_renderer_current_action_compact_with_single_tool_call_dict():
)
message = ContextItemRenderer().render([action])[0]
text = message["content"][0]["text"]
- assert "" in text and "" not in text
- assert '"name": "execute"' in unescape(text) and '"code": "print(1)"' in unescape(text)
+ assert "input:\n" in text and "" not in text
+ assert 'tool: execute' in text and '"code": "print(1)"' in unescape(text)
def test_renderer_current_action_preserves_string_tool_arguments():
@@ -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
@@ -402,9 +402,9 @@ def test_renderer_current_action_without_step_number():
text = "\n".join(m["content"][0]["text"] for m in ContextItemRenderer().render([action]))
assert "index:" not in text
- assert "" in text and '"name": "search"' in unescape(text)
+ assert 'tool: search\ninput:\n' in text
assert "" not in text
- assert "done " in text
+ assert "recorded_result:\ndone" in text
def test_renderer_summary_legacy_dict_renders_markdown():
@@ -478,7 +478,7 @@ def test_renderer_current_action_compact_non_standard_tool_calls():
)
message = ContextItemRenderer().render([action])[0]
text = message["content"][0]["text"]
- assert "" in text
+ assert "tool_record:" in text
def test_renderer_current_action_compact_list_tool_call_non_dict():
@@ -493,7 +493,7 @@ def test_renderer_current_action_compact_list_tool_call_non_dict():
)
message = ContextItemRenderer().render([action])[0]
text = message["content"][0]["text"]
- assert "" in text
+ assert "tool_record:" in text
def test_context_manager_management_and_diagnostic_helpers():
diff --git a/test/sdk/core/agents/test_context_manager_assembly.py b/test/sdk/core/agents/test_context_manager_assembly.py
index 1ae00ad618..bc0ba96204 100644
--- a/test/sdk/core/agents/test_context_manager_assembly.py
+++ b/test/sdk/core/agents/test_context_manager_assembly.py
@@ -2,6 +2,8 @@
"""Focused tests for ContextManager-owned managed assembly."""
from __future__ import annotations
+from html import unescape
+
import pytest
from nexent.core.agents.context import (
ContextItemInput,
@@ -208,7 +210,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,9 +217,9 @@ def test_context_manager_owns_final_answer_assembly():
"final instruction",
"memory fact",
]
- action_history = _message_text(final.messages[4])
- assert '' in action_history
- assert "work trace " in action_history
+ action_history = _message_text(final.messages[3])
+ assert '' in action_history
+ assert "recorded_result:\nwork trace" in action_history
assert "Calling tools:" not in action_history
assert "Observation:" not in action_history
assert final.evidence.purpose == "final_answer"
@@ -292,12 +293,12 @@ def test_completed_skill_read_is_available_to_the_next_model_step():
action_history = next(
_message_text(message)
for message in final.messages
- if '' in _message_text(message)
+ 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)
+ assert 'tool: python_interpreter\ninput:\nskill_content = read_skill_md("demo-skill")' in unescape(action_history)
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
+ assert "Do not copy this record's format as your next response." in action_history
def test_context_manager_attributes_tool_schema_change():
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..bb4b4eca8d 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,
}))
@@ -38,17 +40,17 @@ def text(messages):
@pytest.mark.parametrize('value', ['0', 'False', '[]', ''])
def test_ac002_values_are_not_dropped_or_given_guessed_types(value):
rendered = text(ContextItemRenderer().render([item({'RETURN_VALUE': value})]))
- assert f'{value} ' in rendered
+ assert f'recorded_result:\n{value}' in rendered
assert 'type=' not in rendered
assert rendered.count('print(result)') == 1
- assert 'completed_action_history' not in rendered
+ assert rendered.count('', 'ERROR':'timeout & retry'})]))
- assert 'partial <data> ' in rendered
- assert 'timeout & retry ' in rendered
+ assert 'outcome:\npartial <data>' in rendered
+ assert 'recorded_error:\ntimeout & retry' in rendered
def test_ac001_compact_replaces_output_not_appends():
@@ -295,7 +297,7 @@ def test_ac003_runtime_result_contract_requires_actual_execution_before_final_an
assert 'Only the runtime emits CODE_EXEC_RESULT, SUMMARIED_RESULT and OUTPUT_REF tags' in contract
assert 'do not generate these records yourself' in contract
assert 'wait for the actual execution result' in contract
- assert 'final answer in a separate response' in contract
+ assert 'the next response must confirm or revise' in contract
def test_ac003_ac015_finalize_registers_partial_error_once_and_not_final_answers():
diff --git a/test/sdk/core/agents/test_execution_history_review_regressions.py b/test/sdk/core/agents/test_execution_history_review_regressions.py
index 4a0a242d57..78b63b1519 100644
--- a/test/sdk/core/agents/test_execution_history_review_regressions.py
+++ b/test/sdk/core/agents/test_execution_history_review_regressions.py
@@ -29,6 +29,20 @@
from nexent.core.utils.observer import MessageObserver
+def assert_legacy_calls(text, calls):
+ """Check names, argument values and order in the restored history format."""
+ remaining = unescape(text)
+ for call in calls:
+ if isinstance(call, dict) and 'name' in call:
+ value = call.get('arguments', {})
+ arguments = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)
+ expected = f"tool: {call['name']}\ninput:\n{arguments}"
+ else:
+ expected = 'tool_record: ' + (call if isinstance(call, str) else json.dumps(call, ensure_ascii=False))
+ assert expected in remaining
+ remaining = remaining.split(expected, 1)[1]
+
+
@pytest.mark.parametrize("arguments", [{"q": "foo"}, "foo", {}, None])
@pytest.mark.parametrize("single", [False, True])
def test_ac_r5_ordinary_legacy_calls_preserve_names_and_arguments(arguments, single):
@@ -42,22 +56,27 @@ def test_ac_r5_ordinary_legacy_calls_preserve_names_and_arguments(arguments, sin
original = deepcopy(item.content)
messages = ContextItemRenderer().render([item])
texts = [m["content"][0]["text"] for m in messages]
- assert [json.loads(unescape(t.removeprefix("").removesuffix(" ")))
- for t in texts[:-1]] == calls
+ assert len(texts) == 1
+ assert_legacy_calls(texts[0], calls)
assert all("" not in t for t in texts)
- assert "" in texts[-1] and "known error " in texts[-1]
+ assert "outcome:\ncompleted" in texts[-1] and "recorded_error:\nknown 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_legacy_calls(text, [
+ {"name": "python_interpreter", "arguments": arguments}
+ ])
+ assert "recorded_result:\ndone" in text
+
def test_ac_r5_mixed_legacy_calls_keep_order_and_unknown_records_as_data():
@@ -70,11 +89,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_legacy_calls(texts[0], calls)
assert item.content == original
@@ -97,7 +114,7 @@ def generate(self, messages, **kwargs):
prompt_templates=templates, context_runtime=runtime, max_steps=1, verbosity_level=0)
assert agent.run("Finish") == "done"
assert len(captured) == 1 and "search" in captured[0] and "archive" in captured[0]
- assert "" in captured[0] and "" not in captured[0]
+ assert "tool: search" in captured[0] and "" not in captured[0]
@pytest.mark.parametrize("passthrough", [False, True])
@@ -110,9 +127,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 "tool: search" in str(messages)
+ assert "recorded_result:\ndone" in messages[0]['content'][0]['text']
@pytest.mark.parametrize("history_enabled", [False, True])
@@ -279,9 +300,9 @@ def test_no_record_action_compacts_only_above_threshold_without_fake_reference(h
model=model, memory=memory, current_run_start_idx=0,
run_context=manager.prepare_run_context(memory, "", []),
)
- rendered = str(result.messages)
+ rendered = '\n'.join(part['text'] for message in result.messages for part in message['content'])
assert "head" in rendered and "tail" in rendered
- assert "known failure " in rendered
+ assert "recorded_error:\nknown failure" in rendered
assert "OUTPUT_REF" not in rendered and "SUMMARIED_RESULT" not in rendered
assert rendered.count("print(data)") == 1
assert step.observations == original
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..8ff486baf5 100644
--- a/test/sdk/core/agents/test_history_context_runtime.py
+++ b/test/sdk/core/agents/test_history_context_runtime.py
@@ -272,12 +272,12 @@ def test_projects_planning_and_multiple_actions_in_stable_run_order(monkeypatch)
"CURRENT TASK", "CURRENT PLAN", "observation 1", "observation 2",
)]
assert positions == sorted(positions)
- assert '' in rendered
+ assert '' in rendered
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,16 +317,16 @@ 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 "" in rendered
- assert "done " in rendered
+ assert "" not in rendered
+ assert "print(" in rendered and "ok" in rendered
+ assert "outcome:\nok" in rendered
+ assert "recorded_result:\ndone" 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 "" in summary_rendered
+ assert "" not in summary_rendered
+ assert "outcome:\nok" 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_history_rendering_v282.py b/test/sdk/core/agents/test_history_rendering_v282.py
new file mode 100644
index 0000000000..79c604c37f
--- /dev/null
+++ b/test/sdk/core/agents/test_history_rendering_v282.py
@@ -0,0 +1,183 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""Legacy action presentation with canonical, non-duplicated runtime results."""
+
+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.utils.observer import MessageObserver
+
+
+def render(fields=None, **content):
+ if fields is not None:
+ content['execution_fields'] = fields
+ item = ContextItem.from_input(ContextItemInput(
+ id='step', type='current_action', content=content,
+ metadata={'language': 'en'},
+ ))
+ before = deepcopy(item.content)
+ messages = ContextItemRenderer().render([item])
+ assert len(messages) == 1 and messages[0]['role'] == 'assistant'
+ assert item.content == before
+ return messages[0]['content'][0]['text']
+
+
+def test_v282_action_shape_keeps_actual_call_and_excludes_model_narration():
+ text = render({'STDOUT': 'REAL_OUTPUT'}, step_number=7,
+ assistant_output='FAKE_SUCCESS invented()',
+ tool_calls=[{'name': 'python_interpreter', 'arguments': 'print(actual())'}])
+ assert text.startswith('')
+ assert '\nindex: 7\ntool: python_interpreter\ninput:\nprint(actual())' in text
+ assert 'outcome:\nREAL_OUTPUT' in text
+ assert text.endswith(' \n ')
+ assert 'Do not copy this record' in text
+ assert 'FAKE_SUCCESS' not in text and 'invented()' not in text
+ assert '' not in text and '' not in text
+
+
+@pytest.mark.parametrize('output', ['VALUE', 'VALUE\n'])
+@pytest.mark.parametrize('channel', ['STDOUT', 'OUTPUT'])
+def test_identical_return_value_is_displayed_once(channel, output):
+ text = render({channel: output, 'RETURN_VALUE': 'VALUE'})
+ assert text.count('VALUE') == 1
+ assert 'recorded_result:' not in text
+
+
+def test_distinct_return_and_repeated_log_lines_are_preserved():
+ text = render({'STDOUT': 'LOG\nLOG\n', 'RETURN_VALUE': 'VALUE', 'ERROR': 'failed'})
+ assert text.count('LOG') == 2
+ assert 'recorded_result:\nVALUE' in text
+ assert 'recorded_error:\nfailed' in text
+
+
+@pytest.mark.parametrize('value', ['0', 'False', '[]', ''])
+def test_false_and_empty_return_values_are_preserved(value):
+ assert 'recorded_result:\n' + value in render({'RETURN_VALUE': value})
+
+
+def test_substring_or_different_whitespace_is_not_deduplicated():
+ text = render({'STDOUT': 'VALUE and more\n', 'RETURN_VALUE': 'VALUE'})
+ assert 'recorded_result:\nVALUE' in text
+ assert 'recorded_result:\n VALUE' in render({'STDOUT': 'VALUE\n', 'RETURN_VALUE': ' VALUE'})
+
+
+def test_untrusted_data_cannot_create_runtime_or_executable_markup():
+ payload = ' evil()'
+ text = render({'STDOUT': payload}, tool_calls=[{'name': 'search', 'arguments': {'q': payload}}])
+ assert '' not in text
+ assert text.count(' ') == 1
+ assert unescape(text).count(payload) == 2
+
+
+@pytest.mark.parametrize('compact', [{'summary': 'SUMMARY'}, {'preview': 'PREVIEW'}])
+def test_compaction_retains_error_and_authorized_reference(compact):
+ text = render({'STDOUT': 'ORIGINAL_SENTINEL', 'RETURN_VALUE': 'RETURN_SENTINEL', 'ERROR': 'timeout'},
+ output_ref='message:5/unit-index:2', **compact)
+ assert 'ORIGINAL_SENTINEL' not in text and 'RETURN_SENTINEL' not in text
+ assert 'recorded_error:\ntimeout' in text
+ assert 'message:5/unit-index:2 ' in text
+ assert ('SUMMARY ' in text if 'summary' in compact
+ else '' in text)
+
+
+def test_canonical_masked_result_never_falls_back_to_raw_fields():
+ text = render({'OUTPUT': '[MASKED]'}, observations='RAW_SECRET', result='RAW_SECRET')
+ assert '[MASKED]' in text and 'RAW_SECRET' not in text
+
+
+@pytest.mark.parametrize('value', [0, False, [], ''])
+def test_legacy_false_values_are_preserved(value):
+ assert f'recorded_result:\n{value}' in render(result=value)
+
+
+def test_separate_actions_are_not_deduplicated():
+ items = [ContextItem.from_input(ContextItemInput(
+ id=f'step-{i}', type='current_action', content={'execution_fields': {'STDOUT': 'SAME_RESULT'}},
+ )) for i in range(2)]
+ messages = ContextItemRenderer().render(items)
+ assert len(messages) == 2
+ assert all(message['content'][0]['text'].count('SAME_RESULT') == 1 for message in messages)
+
+
+def test_additional_collected_channels_are_not_dropped():
+ assert 'outcome:\nSTDERR:\nwarning <data>' in render({'STDERR': 'warning '})
+
+
+def test_passthrough_messages_remain_unchanged():
+ messages = [{'role': 'assistant', 'content': [{'type': 'text', 'text': 'external'}]}]
+ item = ContextItem.from_input(ContextItemInput(id='external', type='current_action', content={'messages': messages}))
+ assert ContextItemRenderer().render([item]) == messages
+
+
+@pytest.mark.parametrize('strict', [False, True])
+def test_actual_call_executes_once_and_next_request_has_one_real_result(strict):
+ calls = []
+ requests = []
+
+ class ActualTool(Tool):
+ name = 'actual'
+ description = 'Return the real result.'
+ inputs = {}
+ output_type = 'string'
+
+ def forward(self):
+ calls.append('actual')
+ return 'ACTUAL_RESULT_SENTINEL'
+
+ class SequencedModel(Model):
+ def generate(self, messages, **kwargs):
+ requests.append(json.dumps(messages, default=str))
+ if len(requests) == 1:
+ return ChatMessage(role='assistant', content=(
+ 'MODEL_NARRATION_SENTINEL\nvalue = actual()\nprint(value)\nvalue'
+ 'FAKE_SUCCESS_SENTINEL '
+ ))
+ assert requests[-1].count('ACTUAL_RESULT_SENTINEL') == 1
+ assert 'FAKE_SUCCESS_SENTINEL' not in requests[-1]
+ assert 'MODEL_NARRATION_SENTINEL' not in requests[-1]
+ assert '' in requests[-1] and 'index: 1' in requests[-1]
+ assert 'tool: python_interpreter' in requests[-1]
+ assert '' not in requests[-1]
+ return ChatMessage(role='assistant', content='Confirmed actual result ')
+
+ templates = deepcopy(EMPTY_PROMPT_TEMPLATES)
+ templates['system_prompt'] = 'Run the tool and report only its actual result.'
+ agent = CoreAgent(
+ observer=MessageObserver(lang='en'), model=SequencedModel(model_id='controlled'),
+ tools=[ActualTool()], prompt_templates=templates, verbosity_level=0, max_steps=3,
+ context_runtime=ManagedContextRuntime(ContextManager(), []), enable_protocol_repair_retry=strict,
+ )
+ assert str(agent.run('Execute the actual tool once.')) == 'Confirmed actual result'
+ assert calls == ['actual'] and len(requests) == 2
+
+
+@pytest.mark.parametrize('value', ['VALUE', 'line one\nline two\n', 'quote " and \\ literal', '0', 'False', ''])
+@pytest.mark.parametrize('print_newline', ['', '\n'])
+def test_docker_string_repr_return_is_not_rendered_twice(value, print_newline):
+ fields = {'OUTPUT': value + print_newline, 'RETURN_VALUE': repr(value)}
+ text = render(fields, tool_calls=[{'name': 'python_interpreter', 'arguments': 'print(value)\nvalue'}])
+ assert 'recorded_result:' not in text
+ assert fields == {'OUTPUT': value + print_newline, 'RETURN_VALUE': repr(value)}
+
+
+@pytest.mark.parametrize('channel,tool,output,value', [
+ ('STDOUT', 'python_interpreter', 'VALUE\n', "'VALUE'"),
+ ('OUTPUT', 'search', 'VALUE\n', "'VALUE'"),
+ ('OUTPUT', 'python_interpreter', 'VALUE\nVALUE\n', "'VALUE'"),
+ ('OUTPUT', 'python_interpreter', 'VALUE\n', "'DIFFERENT'"),
+ ('OUTPUT', 'python_interpreter', 'VALUE\n', "'unterminated"),
+ ('OUTPUT', 'python_interpreter', 'VALUE\n', "str('VALUE')"),
+])
+def test_repr_dedup_preserves_other_channels_tools_and_distinct_values(channel, tool, output, value):
+ text = render({channel: output, 'RETURN_VALUE': value}, tool_calls=[{'name': tool}])
+ assert 'recorded_result:' in text
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..d05fdd57f8
--- /dev/null
+++ b/test/sdk/core/agents/test_output_protocol_patch.py
@@ -0,0 +1,136 @@
+# 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
+
+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 and '' not in text
+ assert 'recorded_result:\nFalse' in text and 'recorded_error:\nactual failure' in text
+ assert 'tool: python_interpreter\ninput:\nactual()' in text
+
+
+@pytest.mark.parametrize('strict', [False, True])
+def test_ac004_mixed_stream_executes_once_and_continues_with_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'
+
+ observer = MessageObserver(lang='en')
+
+ class RuntimeModel(Model):
+ def generate(self, messages, **kwargs):
+ requests.append(json.dumps(messages, default=str))
+ if len(requests) == 1:
+ content = 'print(actual())FAKE_SUCCESS '
+ observer.begin_model_attempt('mixed-action', 1)
+ for offset in range(0, len(content), 3):
+ observer.add_model_new_token(content[offset:offset + 3])
+ 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') == 'FAKE_SUCCESS'
+ assert calls == []
+ return ChatMessage(role='assistant', content=content)
+ 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=observer, 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/models/test_implicit_output_cap.py b/test/sdk/core/models/test_implicit_output_cap.py
new file mode 100644
index 0000000000..e671eecdaa
--- /dev/null
+++ b/test/sdk/core/models/test_implicit_output_cap.py
@@ -0,0 +1,222 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""Provider output caps must be explicit rather than context-planning fallbacks."""
+
+import json
+from unittest.mock import MagicMock
+
+import httpx
+import pytest
+from openai import BadRequestError
+
+from nexent.core.models.capacity_budget import (
+ CallerMaxTokensOverrideForbidden,
+ CapacityReservePolicy,
+ ContextBudgetCalculator,
+ ContextBudgetCapacityMismatch,
+ RequestBudgetOverrides,
+)
+from nexent.core.models.capacity_resolver import resolve_capacity
+from nexent.core.models.openai_llm import OpenAIModel
+
+
+def contract(default=None, source="model_default"):
+ overrides = {"context_window_tokens": 32768, "max_output_tokens": 16384}
+ if default is not None:
+ overrides["default_output_reserve_tokens"] = default
+ capacity = resolve_capacity(model_id="test-model", provider="test", operator_overrides=overrides,
+ capability_profiles={})
+ kwargs = {}
+ if source == "agent":
+ kwargs.update(requested_output_tokens=4096, output_reserve_source="agent")
+ if source == "request":
+ kwargs["request_overrides"] = RequestBudgetOverrides(requested_output_tokens=8192)
+ budget = ContextBudgetCalculator().calculate_context_budget(
+ capacity_snapshot=capacity, reserve_policy=CapacityReservePolicy(), **kwargs
+ )
+ projection = {"capacity_fingerprint": capacity.fingerprint, "provider": capacity.provider,
+ "model_name": capacity.model_name, "default_output_reserve_tokens": default}
+ return capacity, budget, projection
+
+
+def model():
+ result = OpenAIModel.__new__(OpenAIModel)
+ result.client = MagicMock()
+ result.model_id = "test-model"
+ result.explicit_sampling = ()
+ return result
+
+
+@pytest.mark.parametrize("default,source,caller,expected", [
+ (None, "model_default", None, None),
+ (4096, "model_default", None, 4096),
+ (8192, "model_default", None, 8192),
+ (None, "agent", None, 4096),
+ (None, "request", None, 8192),
+ (None, "model_default", 4096, 4096),
+])
+def test_dispatch_omits_only_implicit_caps(default, source, caller, expected):
+ capacity, budget, projection = contract(default, source)
+ client = model()
+ before = budget.model_dump()
+ supplied = {} if caller is None else {"max_tokens": caller}
+ client._dispatch_chat_completion(context_budget_snapshot=budget, capacity_snapshot=projection,
+ messages=[], **supplied)
+ sent = client.client.chat.completions.create.call_args.kwargs
+ if expected is None:
+ assert "max_tokens" not in sent
+ else:
+ assert sent["max_tokens"] == expected
+ assert budget.model_dump() == before
+ assert budget.effective_input_limit_tokens == 32768 - budget.requested_output_tokens
+ assert budget.compaction_target_tokens < budget.compaction_trigger_threshold_tokens
+ assert budget.compaction_trigger_threshold_tokens < budget.effective_input_limit_tokens
+ assert budget.w1_fingerprint == capacity.fingerprint
+
+
+@pytest.mark.parametrize("projection", [None, {}, {"provider": "test", "model_name": "test-model"}])
+def test_legacy_missing_reserve_metadata_keeps_snapshot_cap(projection):
+ _, budget, _ = contract()
+ client = model()
+ client._dispatch_chat_completion(context_budget_snapshot=budget, capacity_snapshot=projection, messages=[])
+ assert client.client.chat.completions.create.call_args.kwargs["max_tokens"] == 4096
+
+
+@pytest.mark.parametrize("failure", ["caller", "fingerprint", "model"])
+def test_omission_does_not_bypass_trusted_budget_checks(failure):
+ _, budget, projection = contract()
+ kwargs = {}
+ if failure == "caller":
+ kwargs["max_tokens"] = 8192
+ error = CallerMaxTokensOverrideForbidden
+ elif failure == "model":
+ projection["model_name"] = "other-model"
+ error = ContextBudgetCapacityMismatch
+ else:
+ projection["capacity_fingerprint"] = "other-fingerprint"
+ error = ContextBudgetCapacityMismatch
+ client = model()
+ with pytest.raises(error):
+ client._dispatch_chat_completion(context_budget_snapshot=budget, capacity_snapshot=projection,
+ messages=[], **kwargs)
+ client.client.chat.completions.create.assert_not_called()
+
+
+def test_sampling_fallback_does_not_restore_implicit_cap():
+ _, budget, projection = contract()
+ client = model()
+ request = httpx.Request("POST", "http://fixture/v1/chat/completions")
+ response = httpx.Response(400, request=request)
+ client.client.chat.completions.create.side_effect = [
+ BadRequestError("unsupported temperature", response=response, body={}), "ok"
+ ]
+ assert client._dispatch_chat_completion(context_budget_snapshot=budget, capacity_snapshot=projection,
+ messages=[], temperature=0.2) == "ok"
+ assert client.client.chat.completions.create.call_count == 2
+ assert all("max_tokens" not in call.kwargs
+ for call in client.client.chat.completions.create.call_args_list)
+
+
+@pytest.mark.parametrize("outcome", ["success", "empty_length", "overflow_then_success", "overflow_exhausted"])
+def test_streaming_and_overflow_recovery_without_wire_cap(outcome):
+ from openai import OpenAI
+
+ from nexent.core.context_runtime.contracts import FinalContext
+ from nexent.core.models.retry import ModelInvocationTerminalError
+ from nexent.core.utils.observer import MessageObserver
+
+ _, budget, projection = contract()
+ requests = []
+ rebuilds = []
+
+ def provider(request):
+ requests.append(json.loads(request.content))
+ if outcome == "overflow_exhausted" or (outcome == "overflow_then_success" and len(requests) == 1):
+ return httpx.Response(400, json={"error": {
+ "message": "maximum context length exceeded", "type": "invalid_request_error",
+ "code": "context_length_exceeded",
+ }})
+ empty = outcome == "empty_length"
+ delta = {"role": "assistant", "reasoning_content": "thinking"} if empty else {
+ "role": "assistant", "content": "actual reply "
+ }
+ chunk = {"id": "fixture", "object": "chat.completion.chunk", "created": 0,
+ "model": "test-model", "choices": [{"index": 0, "delta": delta,
+ "finish_reason": "length" if empty else "stop"}],
+ "usage": {"prompt_tokens": 16, "completion_tokens": 4096 if empty else 10,
+ "total_tokens": 4112 if empty else 26}}
+ return httpx.Response(200, headers={"content-type": "text/event-stream"},
+ text="data: " + json.dumps(chunk) + "\n\ndata: [DONE]\n\n")
+
+ def rebuild():
+ rebuilds.append(True)
+ return FinalContext(messages=[{"role": "user", "content": "rebuilt context"}])
+
+ instance = OpenAIModel(observer=MessageObserver(lang="en"), model_id="test-model",
+ api_base="http://provider.invalid/v1", api_key="fixture-only",
+ capacity_snapshot=projection, context_budget_snapshot=budget)
+ original_client = instance.client
+ with OpenAI(api_key="fixture-only", base_url="http://provider.invalid/v1", max_retries=0,
+ http_client=httpx.Client(transport=httpx.MockTransport(provider))) as client:
+ instance.client = client
+ if outcome in {"empty_length", "overflow_exhausted"}:
+ with pytest.raises(ModelInvocationTerminalError):
+ instance([{"role": "user", "content": "original context"}], context_rebuild=rebuild)
+ else:
+ result = instance([{"role": "user", "content": "original context"}], context_rebuild=rebuild)
+ assert "actual reply" in result.content
+ original_client.close()
+ assert all("max_tokens" not in request for request in requests)
+ assert len(requests) == {"success": 1, "empty_length": 1,
+ "overflow_then_success": 2, "overflow_exhausted": 3}[outcome]
+ assert len(rebuilds) == {"success": 0, "empty_length": 0,
+ "overflow_then_success": 1, "overflow_exhausted": 2}[outcome]
+ if rebuilds:
+ assert all("rebuilt context" in str(request["messages"]) for request in requests[1:])
+ assert instance.context_budget_snapshot == budget
+ if outcome == "empty_length":
+ assert instance.last_response_diagnostics["requested_output_tokens"] is None
+ assert instance.last_response_diagnostics["reasoning_only_budget_exhausted"] is True
+
+
+@pytest.mark.parametrize("default", [None, 4096])
+def test_resolved_budget_keeps_history_compaction_and_final_fit(default):
+ from types import SimpleNamespace
+
+ from smolagents.memory import AgentMemory
+
+ from nexent.core.agents.context import ContextItemInput, ContextManager, ContextManagerConfig
+ from nexent.core.agents.context.policy import ContextPolicy, ContextProcessingMode, PolicyLayers
+
+ _, budget, projection = contract(default)
+ manager = ContextManager(ContextManagerConfig(
+ context_window_tokens=32768,
+ effective_input_limit_tokens=budget.effective_input_limit_tokens,
+ compaction_trigger_threshold_tokens=budget.compaction_trigger_threshold_tokens,
+ compaction_target_tokens=budget.compaction_target_tokens,
+ policy_layers=PolicyLayers(request=ContextPolicy(processing_mode=ContextProcessingMode.ADAPTIVE_COMPACT)),
+ ))
+ item = ContextItemInput(id="turn:1", type="conversation_turn", content={
+ "user_message": "old question " * 15000, "assistant_final_answer": "old answer " * 15000,
+ "user_message_id": 1, "assistant_message_id": 2,
+ })
+ memory = AgentMemory(system_prompt="system")
+ run = manager.prepare_run_context(memory, "", items=[item])
+ summary = "# Compact Result of History\n" + "\n".join(
+ "## " + heading + "\npreserved history" for heading in [
+ "Task Overview", "Completed Work", "Key Decisions", "Unresolved Issues",
+ "Pending Items", "Next Steps", "Context To Preserve",
+ ]
+ )
+ summarize = MagicMock(return_value=SimpleNamespace(content=summary, token_usage=None))
+ final = manager.assemble_final_context(model=summarize, memory=memory, current_run_start_idx=0, run_context=run)
+ assert final.evidence.history_compression_triggered is True
+ assert final.evidence.new_summary_coverage is not None
+ assert final.evidence.final_token_estimate <= budget.effective_input_limit_tokens
+ assert not final.evidence.exceeds_effective_input_limit
+ assert "preserved history" in str(final.messages)
+ client = model()
+ client._dispatch_chat_completion(context_budget_snapshot=budget, capacity_snapshot=projection,
+ messages=final.messages)
+ sent = client.client.chat.completions.create.call_args.kwargs
+ assert ("max_tokens" in sent) is (default is not None)
+ assert len(sent["messages"]) > 0
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..994ba99aef
--- /dev/null
+++ b/test/sdk/core/utils/test_observer_mixed_answer.py
@@ -0,0 +1,37 @@
+# Copyright (c) 2025-2026 Huawei Technologies Co., Ltd. All rights reserved.
+"""Mixed action/final responses retain ordinary streaming and attempt isolation."""
+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_streams_without_waiting_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 ''.join(event.get('content', '') for event in events if event.get('type') == 'final_answer') == 'UNCONFIRMED'
+ observer.commit_model_attempt('action', 1)
+ observer.begin_model_attempt('next-response', 1)
+ observer.add_model_new_token('CONFIRMED ')
+ observer.flush_remaining_tokens()
+ events = [json.loads(value) for value in observer.message_query]
+ for attempt, expected in [('action', 'UNCONFIRMED'), ('next-response', 'CONFIRMED')]:
+ assert ''.join(event.get('content', '') for event in events
+ if event.get('type') == 'final_answer' and event.get('attempt_id') == attempt) == expected
+
+
+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..2c3b245caf
--- /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[3]
+ 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