diff --git a/artemis/llm/argument_fingerprint.py b/artemis/llm/argument_fingerprint.py new file mode 100644 index 0000000..2630a5d --- /dev/null +++ b/artemis/llm/argument_fingerprint.py @@ -0,0 +1,53 @@ +"""Bounded fingerprints of received argument deltas, without retaining text.""" + +import hashlib + + +PREFIX_BYTES = 4096 +MAX_PREFIXES = 16 +MAX_TRACKED_TOOLS = 64 + + +class ArgumentFingerprint: + def __init__(self): + self.characters = 0 + self.bytes = 0 + self.deltas = 0 + self.encoding_valid = True + self.digest = hashlib.sha256() + self.prefixes: list[dict] = [] + + def observe(self, arguments: str) -> None: + self.characters += len(arguments) + if not arguments: + return + self.deltas += 1 + if not self.encoding_valid: + return + try: + encoded = arguments.encode("utf-8") + except UnicodeEncodeError: + # Diagnostics must not repair, reject or otherwise alter the stream. + self.encoding_valid = False + return + offset = 0 + while offset < len(encoded) and len(self.prefixes) < MAX_PREFIXES: + size = min(PREFIX_BYTES - self.bytes % PREFIX_BYTES, len(encoded) - offset) + self.digest.update(encoded[offset : offset + size]) + self.bytes += size + offset += size + if self.bytes % PREFIX_BYTES == 0: + self.prefixes.append({"bytes": self.bytes, "sha256": self.digest.hexdigest()}) + if offset < len(encoded): + self.digest.update(encoded[offset:]) + self.bytes += len(encoded) - offset + + def snapshot(self) -> dict: + return { + "argument_chars": self.characters, + "argument_bytes": self.bytes if self.encoding_valid else None, + "argument_sha256": self.digest.hexdigest() if self.encoding_valid else None, + "argument_deltas": self.deltas, + "argument_prefixes": [dict(prefix) for prefix in self.prefixes], + "argument_encoding_valid": self.encoding_valid, + } diff --git a/artemis/llm/diagnostics.py b/artemis/llm/diagnostics.py index 36add48..fc4716c 100644 --- a/artemis/llm/diagnostics.py +++ b/artemis/llm/diagnostics.py @@ -13,6 +13,7 @@ from uuid import uuid4 from PIL import Image from artemis.data_engine.context_vars import CURRENT_NODE_NAME, CURRENT_TRACE_ID +from artemis.llm.argument_fingerprint import ArgumentFingerprint, MAX_TRACKED_TOOLS from artemis.llm.final_review import ACTIVE_FINAL_REVIEW, FinalReviewBudgetExceeded from artemis.llm.reliability import LLMExhaustedError from artemis.llm.usage import usage_from_message @@ -107,6 +108,7 @@ class RequestAttempt: self.clock = time.monotonic() self.last_publish = self.clock self.tools: dict[int | str, dict] = {} + self.tool_fingerprints: dict[int | str, ArgumentFingerprint] = {} self.review = ACTIVE_FINAL_REVIEW.get() if self.review: self.review.request_started(request_id, attempt) @@ -126,6 +128,9 @@ class RequestAttempt: "usage": None, "output_chunks": 0, "tool_argument_chars": 0, + "tool_observation_layer": "langchain_tool_call_chunks", + "argument_fingerprint_version": 1, + "tool_stats_truncated": False, "tools": [], "step_id": str(self.step_id) if self.step_id else None, "parent_trace_id": str(self.parent_id) if self.parent_id else None, @@ -167,9 +172,10 @@ class RequestAttempt: self.data.update( headers_at=time.time(), http_status=response.status_code, - server_request_id=_identifier( - response.headers.get("x-request-id") or response.headers.get("request-id") - ), + server_request_id=_identifier(response.headers.get("x-oneapi-request-id")) + or _identifier(response.headers.get("x-request-id")) + or _identifier(response.headers.get("request-id")), + response_x_request_id=_identifier(response.headers.get("x-request-id")), server_trace_id=_identifier(response.headers.get("x-trace-id")), phase="waiting_output", ) @@ -196,18 +202,31 @@ class RequestAttempt: self.data["usage"] = usage tool_chunks = getattr(chunk, "tool_call_chunks", None) or [] # Responses exposes the same argument delta in content and tool_call_chunks. - # Count only the latter; retain lengths/names, never argument bodies. + # Count only the latter; retain bounded fingerprints, never argument bodies. for tool in tool_chunks: + arguments = tool.get("args") + if isinstance(arguments, str): + self.data["tool_argument_chars"] += len(arguments) index = tool.get("index") - key = index if index is not None else tool.get("id") or "unknown" - stats = self.tools.setdefault(key, {"name": None, "argument_chars": 0}) + call_id = _identifier(tool.get("id")) + key = index if index is not None else call_id or "unknown" + if key not in self.tools and len(self.tools) >= MAX_TRACKED_TOOLS: + self.data["tool_stats_truncated"] = True + continue + stats = self.tools.setdefault(key, {"name": None}) + if call_id: + stats["call_id"] = call_id + if index is not None: + stats["chunk_index"] = index name = tool.get("name") if isinstance(name, str) and re.fullmatch(r"[\w.:-]{1,128}", name): stats["name"] = name - arguments = tool.get("args") + fingerprint = self.tool_fingerprints.get(key) + if fingerprint is None: + fingerprint = self.tool_fingerprints[key] = ArgumentFingerprint() if isinstance(arguments, str): - stats["argument_chars"] += len(arguments) - self.data["tool_argument_chars"] += len(arguments) + fingerprint.observe(arguments) + stats.update(fingerprint.snapshot()) self.data["tools"] = list(self.tools.values()) content = getattr(chunk, "content", None) substantive = isinstance(content, str) and bool(content) @@ -252,7 +271,9 @@ class RequestAttempt: if isinstance(status, int): self.data["http_status"] = status if request_id := _identifier(getattr(error, "request_id", None)): - self.data["server_request_id"] = request_id + self.data["error_request_id"] = request_id + if not self.data.get("server_request_id"): + self.data["server_request_id"] = request_id if category == "timeout": self.data.setdefault("timeout_scope", "transport") if response is not None: @@ -301,6 +322,7 @@ async def observe_request_headers(request): if attempt is None: return request.headers.setdefault("X-Client-Request-Id", attempt.data["request_id"]) + request.headers.setdefault("X-Client-Attempt-Id", attempt.data["attempt_id"]) routing = {} for name in ("thread-id", "thread_id", "session-id", "session_id"): value = request.headers.get(name) @@ -309,9 +331,12 @@ async def observe_request_headers(request): attempt.data.update( request_sent_at=time.time(), client_request_id=_identifier(request.headers.get("X-Client-Request-Id")), + client_attempt_id=_identifier(request.headers.get("X-Client-Attempt-Id")), routing_header_fingerprints=routing, client_request_id_matches=request.headers.get("X-Client-Request-Id") == attempt.data["request_id"], + client_attempt_id_matches=request.headers.get("X-Client-Attempt-Id") + == attempt.data["attempt_id"], ) attempt.publish() diff --git a/artemis/llm/reliability.py b/artemis/llm/reliability.py index 4e535fb..507f227 100644 --- a/artemis/llm/reliability.py +++ b/artemis/llm/reliability.py @@ -153,7 +153,7 @@ class LLMPermanentError(LLMCallError): class LLMExhaustedError(LLMCallError): - """Retryable failures exhausted every attempt and the pause deadline.""" + """Retryable failures exhausted attempts and any applicable pause deadline.""" # --------------------------------------------------------------------------- diff --git a/artemis/llm/stream_guard.py b/artemis/llm/stream_guard.py index a60af3e..3aeb7f4 100644 --- a/artemis/llm/stream_guard.py +++ b/artemis/llm/stream_guard.py @@ -23,7 +23,7 @@ class ToolArgumentOverflow(ValueError): "limit_chars": SMALL_ACTION_ARGUMENT_LIMIT, } super().__init__( - f"模型生成的 {name} 参数异常膨胀（{characters} 字符，" + f"收到的 {name} 参数流超出安全限制（{characters} 字符，" f"保护上限 {SMALL_ACTION_ARGUMENT_LIMIT}）；已关闭输出流，未执行本次动作。" ) diff --git a/artemis/services/llm.py b/artemis/services/llm.py index 2c9d5da..f06c25c 100644 --- a/artemis/services/llm.py +++ b/artemis/services/llm.py @@ -381,7 +381,7 @@ async def _run_with_recovery[T]( immediately: retrying cannot help and pausing would hang the task. - When retryable attempts are exhausted: if a fallback model is waiting (with_fallback), raise LLMExhaustedError so it takes over immediately; - final acceptance also raises instead of pausing an already-finished task. + final acceptance and invalid output also raise instead of globally pausing. Other calls pause (bounded by settings.LLM_PAUSE_TIMEOUT_SECONDS) and retry the same request on resume. """ @@ -470,7 +470,7 @@ async def _run_with_recovery[T]( ) if failure.category is FailureCategory.INVALID_OUTPUT: # One fresh response can recover a generation loop. Repeated - # runaway output must reach the existing fallback/pause path, + # runaway output must reach fallback or fail this request, # even when ordinary network retries are configured higher. max_attempts = min( max_attempts, total_attempts + max(0, policy.max_attempts - attempt) @@ -505,7 +505,11 @@ async def _run_with_recovery[T]( ACTIVE_ATTEMPT.reset(observation_token) # Retryable attempts exhausted. - if _FALLBACK_AVAILABLE.get() or ACTIVE_FINAL_REVIEW.get() is not None: + if ( + _FALLBACK_AVAILABLE.get() + or ACTIVE_FINAL_REVIEW.get() is not None + or last_failure.category is FailureCategory.INVALID_OUTPUT + ): if failed_endpoints is not None and endpoint_key: failed_endpoints[endpoint_key] = last_error _record_llm_event( @@ -514,7 +518,11 @@ async def _run_with_recovery[T]( "error": str(last_error)[:1000], "category": last_failure.category.value, "retryable": True, - "handover": "fallback" if _FALLBACK_AVAILABLE.get() else "final_review_end", + "handover": "fallback" + if _FALLBACK_AVAILABLE.get() + else "final_review_end" + if ACTIVE_FINAL_REVIEW.get() is not None + else "task_failed", }, status="failed", ) @@ -1253,7 +1261,7 @@ async def _with_fallback[T]( ) # The contextvar must be reset BEFORE fallback_call runs: the fallback has - # no further fallback behind it, so its own exhaustion should pause. + # no further fallback behind it, so apply its category's terminal policy. fallback_token = _FALLBACK_AVAILABLE.set(True) try: result = await main_call() diff --git a/docs/llm-stream-diagnostics.md b/docs/llm-stream-diagnostics.md new file mode 100644 index 0000000..e4f5540 --- /dev/null +++ b/docs/llm-stream-diagnostics.md @@ -0,0 +1,39 @@ +# 模型流式参数诊断 + +小型内置动作的参数流仍受 16384 字符保护。超限时关闭输出流、丢弃不完整参数，不执行该次动作；一次新响应重试和可用备用端点耗尽后，只结束当前任务，不进入共享的模型暂停。批次仍遵循“失败即停止”、人工暂停和取消设置。网络故障、认证错误和最终验收保留各自的恢复规则。 + +## 请求关联 + +每个模型请求尝试沿用现有 `llm_diagnostic` 记录，不另建日志数据库。 + +| 字段 / 请求头 | 含义 | +| --- | --- | +| `request_id` / `X-Client-Request-Id` | 同一次恢复循环内保持稳定的逻辑请求编号；切换备用调用或恢复暂停后可能建立新的逻辑请求 | +| `attempt_id` / `X-Client-Attempt-Id` | 本次客户端尝试编号；重试时变化，`previous_attempt_id` 关联前次尝试 | +| `client_*_id_matches` | 实际发送的请求头是否与本地编号一致；调用方已指定的头保留原值 | +| `server_request_id` | 首个有效的 `X-Oneapi-Request-Id`、`X-Request-Id` 或 `Request-Id`，按此前后顺序选择 | +| `response_x_request_id` | 响应中的原始 `X-Request-Id`，不推定它来自模型上游 | +| `error_request_id` | SDK 异常带回的编号，不覆盖已经从响应头读取的编号 | +| `response_id` | SDK 实际返回的 Responses 响应编号 | + +编号只接收有长度限制的可打印白名单字符。请求头不控制会话亲和性，不替换 `thread-id` / `session-id`。网关自身的重试应使用独立编号或序号，不能假装是一次新的客户端尝试。 + +## 参数指纹 v1 + +`tool_observation_layer=langchain_tool_call_chunks` 明确表示客户端观测的是 SDK 转换后的参数片段，不是原始 SSE 帧。每工具记录实际可得的 `call_id`、`chunk_index`、工具名，以及： + +- `argument_chars`：参数片段的 Unicode 字符数之和。 +- `argument_bytes` / `argument_sha256`：按原顺序连接片段后，其 UTF-8 字节长度和 SHA256；不解析、重排参数 JSON，也不规范化空白或 Unicode。 +- `argument_deltas`：非空参数片段数，不能直接与网关原始事件数比较。 +- `argument_prefixes`：每 4096 字节的累计 SHA256，最多保留前 16 个。可以在分片大小不同、上游预读或中途取消时比较相同字节位置。 +- `argument_encoding_valid`：遇到不能编码为 UTF-8 的片段时为 false，当前总字节数和总哈希留空，避免伪造可比结果；诊断本身不改变执行流程。 + +最多记录 64 个工具；超过时设置 `tool_stats_truncated=true`，总参数字符计数仍继续。诊断不保留参数正文。哈希仅证明已收到的字节前缀，不能证明输出完整或动作成功；完整性仍由响应终止状态和解析校验决定。 + +跨三端优先使用 `response_id + call_id` 关联。LangChain 的 `chunk_index` 可能因文本或推理子块不同于原始 `output_index`，缺失编号时不能按数组位置猜测。正常的 `arguments.done`、`output_item.done`、`response.completed` 快照不重复计入参数；自定义工具转换、非空初始参数或协议异常需要先核对 SDK 映射，不能仅凭跨层哈希不同认定某一端重复发送。 + +历史日志没有保存的原始帧无法由新增诊断恢复。HTTP 200 只说明响应头已到达；客户端保护后出现的 499 或 Broken pipe 也不能证明最初原因。网关写入本机下游 writer 成功不等于客户端已消费或执行。 + +## 离线验证 + +`tests/unit/test_llm_stream_guard.py` 验证保护、有限重试、相同/不同备用端点，以及不会创建或清除其他请求的全局暂停。`tests/unit/test_request_diagnostics.py` 验证请求编号、超时重试、响应头及单次参数计数。`tests/unit/test_argument_fingerprint.py` 验证不同分片、中文和 emoji、固定字节前缀、有界统计与异常编码。这些测试不调用真实模型或设备。 diff --git a/tests/unit/test_argument_fingerprint.py b/tests/unit/test_argument_fingerprint.py new file mode 100644 index 0000000..cfb1407 --- /dev/null +++ b/tests/unit/test_argument_fingerprint.py @@ -0,0 +1,59 @@ +import hashlib +import json + +import pytest +from langchain_core.messages import AIMessageChunk + +from artemis.llm.argument_fingerprint import ArgumentFingerprint, MAX_PREFIXES, PREFIX_BYTES +from artemis.llm.diagnostics import RequestAttempt + + +@pytest.mark.parametrize("fragment_size", [1, 37, 4096, 100000]) +def test_fingerprints_match_at_byte_offsets_despite_unicode_and_different_fragmentation( + fragment_size, +): + arguments = '{"target":"' + "私人内容🙂" * 8000 + '"}' + expected = arguments.encode("utf-8") + fingerprint = ArgumentFingerprint() + for offset in range(0, len(arguments), fragment_size): + fingerprint.observe(arguments[offset : offset + fragment_size]) + result = fingerprint.snapshot() + assert result["argument_chars"] == len(arguments) + assert result["argument_bytes"] == len(expected) + assert result["argument_sha256"] == hashlib.sha256(expected).hexdigest() + assert result["argument_encoding_valid"] is True + assert result["argument_prefixes"] == [ + {"bytes": size, "sha256": hashlib.sha256(expected[:size]).hexdigest()} + for size in range(PREFIX_BYTES, PREFIX_BYTES * MAX_PREFIXES + 1, PREFIX_BYTES) + ] + assert "私人" not in json.dumps(result, ensure_ascii=False) + + +def test_invalid_utf8_is_marked_without_failing_or_forging_a_digest(): + fingerprint = ArgumentFingerprint() + for value in ["safe", "\ud800", "later"]: + fingerprint.observe(value) + assert fingerprint.snapshot() == { + "argument_chars": 10, + "argument_bytes": None, + "argument_sha256": None, + "argument_deltas": 3, + "argument_prefixes": [], + "argument_encoding_valid": False, + } + + +def test_tool_tracking_is_bounded_but_total_count_keeps_all_received_arguments(): + attempt = RequestAttempt( + None, request_id="bounded", attempt=1, timeout=300, max_retries=0, metadata={} + ) + for index in range(65): + attempt.chunk( + AIMessageChunk( + content=[], + tool_call_chunks=[{"name": "click", "args": "{}", "index": index}], + ) + ) + assert len(attempt.data["tools"]) == len(attempt.tool_fingerprints) == 64 + assert attempt.data["tool_stats_truncated"] is True + assert attempt.data["tool_argument_chars"] == 130 diff --git a/tests/unit/test_llm_response_integrity.py b/tests/unit/test_llm_response_integrity.py index a028968..c95c713 100644 --- a/tests/unit/test_llm_response_integrity.py +++ b/tests/unit/test_llm_response_integrity.py @@ -2,6 +2,7 @@ import asyncio import json +import hashlib from types import SimpleNamespace from unittest.mock import AsyncMock @@ -15,6 +16,7 @@ import pytest_asyncio from artemis.llm.reliability import FailureCategory, LLMExhaustedError from artemis.llm.response_guard import ResponseIntegrityError, ensure_response_complete from artemis.llm.router import ModelEndpoint, ModelFactory +from artemis.llm.diagnostics import RequestAttempt from artemis.services import llm as llm_service from artemis.services.llm import RobustChatModelWrapper, acomplete @@ -195,6 +197,33 @@ async def test_completed_stream_returns_original_valid_arguments(transport_case) assert result.tool_calls[0]["args"] == {"passed": True} +@pytest.mark.asyncio +@pytest.mark.parametrize("transport_case", ["fragmented"], indirect=True) +async def test_real_sdk_fingerprint_counts_received_arguments_once(transport_case): + case = transport_case + arguments = json.dumps({"passed": True, "evidence": "中文🙂" * 1000}, ensure_ascii=False) + case.outputs.append(("completed", arguments)) + observation = RequestAttempt( + None, request_id="sdk-test", attempt=1, timeout=300, max_retries=0, metadata={} + ) + merged = None + async for chunk in case.model.astream([]): + observation.chunk(chunk) + merged = chunk if merged is None else merged + chunk + observation.finish("completed", response=merged) + assert len(observation.data["tools"]) == 1 + tool = observation.data["tools"][0] + assert tool["call_id"] == "call_test" + assert tool["argument_chars"] == len(arguments) + assert tool["argument_bytes"] == len(arguments.encode()) + assert tool["argument_sha256"] == hashlib.sha256(arguments.encode()).hexdigest() + assert tool["argument_prefixes"][0] == { + "bytes": 4096, + "sha256": hashlib.sha256(arguments.encode()[:4096]).hexdigest(), + } + assert tool["argument_deltas"] == 1 + + @pytest.mark.asyncio @pytest.mark.parametrize("transport_case", ["fragmented"], indirect=True) @pytest.mark.parametrize("status", ["completed", None]) diff --git a/tests/unit/test_llm_stream_guard.py b/tests/unit/test_llm_stream_guard.py index 222ba0f..1cef531 100644 --- a/tests/unit/test_llm_stream_guard.py +++ b/tests/unit/test_llm_stream_guard.py @@ -124,7 +124,9 @@ async def test_gateway_closes_bad_stream_discards_partial_and_bounds_retry( with pytest.raises(LLMExhaustedError) as caught: await wrapper.complete([]) assert caught.value.failure.category is FailureCategory.INVALID_OUTPUT - llm._wait_for_resume.assert_awaited_once() + llm._wait_for_resume.assert_not_awaited() + assert not llm.PAUSE_FILE.exists() + assert any(name == "llm_gave_up" and p["handover"] == "task_failed" for name, p in resets) assert model.calls == 2 assert closed == [1, 2] failure = next( @@ -139,3 +141,47 @@ async def test_gateway_closes_bad_stream_discards_partial_and_bounds_retry( assert failure["usage"] is None assert "private" not in json.dumps(records + resets, default=str) assert any(name == "llm_stream_reset" and p["action"] == "discard" for name, p in resets) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("fallback_kind", ["same", "failing", "working"]) +@pytest.mark.parametrize("already_paused", [False, True]) +async def test_invalid_output_fallback_does_not_pause_or_clear_another_request( + fallback_kind, already_paused +): + calls = [] + if already_paused: + llm.PAUSE_FILE.write_text("another request or user pause", encoding="utf-8") + + async def main(): + calls.append("main") + raise ToolArgumentOverflow("click", SMALL_ACTION_ARGUMENT_LIMIT + 1) + + async def fallback(): + calls.append("fallback") + if fallback_kind == "working": + return "valid response" + raise ToolArgumentOverflow("click", SMALL_ACTION_ARGUMENT_LIMIT + 2) + + async def run(): + return await llm.with_fallback( + lambda: llm._run_with_recovery(main, endpoint_key="main", max_retries=5), + lambda: llm._run_with_recovery( + fallback, + endpoint_key="main" if fallback_kind == "same" else "fallback", + max_retries=5, + ), + ) + + if fallback_kind == "working": + assert await run() == "valid response" + else: + with pytest.raises(LLMExhaustedError) as caught: + await run() + assert caught.value.failure.category is FailureCategory.INVALID_OUTPUT + assert calls.count("main") == 2 + assert calls.count("fallback") == {"same": 0, "failing": 2, "working": 1}[fallback_kind] + llm._wait_for_resume.assert_not_awaited() + assert llm.PAUSE_FILE.exists() is already_paused + if already_paused: + assert llm.PAUSE_FILE.read_text(encoding="utf-8") == "another request or user pause" diff --git a/tests/unit/test_request_diagnostics.py b/tests/unit/test_request_diagnostics.py index ef5021c..19f4ba6 100644 --- a/tests/unit/test_request_diagnostics.py +++ b/tests/unit/test_request_diagnostics.py @@ -1,6 +1,7 @@ import asyncio from copy import deepcopy import json +import hashlib from types import SimpleNamespace from uuid import uuid4 @@ -36,6 +37,46 @@ class Engine: pass +@pytest.mark.parametrize( + ("headers", "expected"), + [ + ({"X-Oneapi-Request-Id": "gateway-request"}, "gateway-request"), + ( + {"X-Oneapi-Request-Id": "gateway-request", "X-Request-Id": "provider-request"}, + "gateway-request", + ), + ({"X-Request-Id": "provider-request"}, "provider-request"), + ({"Request-Id": "legacy-request"}, "legacy-request"), + ( + {"X-Oneapi-Request-Id": "invalid header value", "X-Request-Id": "standard-request"}, + "standard-request", + ), + ], +) +def test_gateway_request_id_preserves_the_standard_header_without_assuming_its_source( + headers, expected +): + attempt = RequestAttempt( + Engine(), request_id="local-request", attempt=1, timeout=300, max_retries=1, metadata={} + ) + attempt.headers(httpx.Response(200, headers=headers)) + assert attempt.data["server_request_id"] == expected + assert attempt.data["response_x_request_id"] == headers.get("X-Request-Id") + + +def test_sdk_error_cannot_overwrite_the_observed_gateway_request_id(): + class RequestError(Exception): + request_id = "sdk-request" + + attempt = RequestAttempt( + Engine(), request_id="local-request", attempt=1, timeout=300, max_retries=1, metadata={} + ) + attempt.headers(httpx.Response(400, headers={"X-Oneapi-Request-Id": "gateway-request"})) + attempt.finish("failed", error=RequestError("failed")) + assert attempt.data["server_request_id"] == "gateway-request" + assert attempt.data["error_request_id"] == "sdk-request" + + @pytest.mark.asyncio async def test_request_hook_correlates_api_calls_without_logging_header_values(): engine = Engine() @@ -55,13 +96,18 @@ async def test_request_hook_correlates_api_calls_without_logging_header_values() ) await observe_request_headers(request) assert request.headers["x-client-request-id"] == "local-request" + assert request.headers["x-client-attempt-id"] == str(attempt.trace_id) + assert attempt.data["client_attempt_id_matches"] is True assert attempt.data["client_request_id_matches"] is True assert len(attempt.data["routing_header_fingerprints"]) == 2 assert "private" not in json.dumps(engine.records, default=str) request.headers["X-Client-Request-Id"] = "caller-id" + request.headers["X-Client-Attempt-Id"] = "caller-attempt" await observe_request_headers(request) assert request.headers["x-client-request-id"] == "caller-id" assert attempt.data["client_request_id_matches"] is False + assert request.headers["X-Client-Attempt-Id"] == "caller-attempt" + assert attempt.data["client_attempt_id_matches"] is False finally: ACTIVE_ATTEMPT.reset(token) @@ -75,6 +121,7 @@ async def test_headers_without_body_timeout_then_retry_keeps_evidence(monkeypatc ) llm._ENDPOINT_BREAKER._states.clear() calls = 0 + wire_ids = [] class StalledBody(httpx.AsyncByteStream): async def __aiter__(self): @@ -84,6 +131,9 @@ async def test_headers_without_body_timeout_then_retry_keeps_evidence(monkeypatc async def transport(request): nonlocal calls calls += 1 + wire_ids.append( + (request.headers["X-Client-Request-Id"], request.headers["X-Client-Attempt-Id"]) + ) headers = {"x-request-id": f"server-{calls}", "set-cookie": "secret-cookie"} return httpx.Response( 200, headers=headers, stream=StalledBody() if calls == 1 else httpx.ByteStream(b"ok") @@ -91,7 +141,7 @@ async def test_headers_without_body_timeout_then_retry_keeps_evidence(monkeypatc async with httpx.AsyncClient( transport=httpx.MockTransport(transport), - event_hooks={"response": [observe_response_headers]}, + event_hooks={"request": [observe_request_headers], "response": [observe_response_headers]}, ) as client: class Model: @@ -123,6 +173,10 @@ async def test_headers_without_body_timeout_then_retry_keeps_evidence(monkeypatc assert final["request_id"] == failed["request_id"] assert final["attempt_id"] != failed["attempt_id"] and final["attempt"] == 2 assert final["previous_attempt_id"] == failed["attempt_id"] + assert wire_ids == [ + (failed["request_id"], failed["attempt_id"]), + (final["request_id"], final["attempt_id"]), + ] retry = next(r["payload"] for r in engine.records if r["name"] == "llm_retry") assert retry["request_id"] == failed["request_id"] assert retry["attempt_id"] == failed["attempt_id"] @@ -163,10 +217,26 @@ def test_tool_deltas_are_counted_once_without_retaining_arguments(): attempt.finish("completed", response=combined) assert attempt.data["output_chunks"] == 4 assert attempt.data["tool_argument_chars"] == len('{"secret":"private"}{"seconds":1}') - assert attempt.data["tools"] == [ - {"name": "click", "argument_chars": len('{"secret":"private"}')}, - {"name": "wait", "argument_chars": len('{"seconds":1}')}, - ] + for tool, name, arguments, call_id, index, count in zip( + attempt.data["tools"], + ["click", "wait"], + ['{"secret":"private"}', '{"seconds":1}'], + ["c1", "c2"], + [0, 1], + [2, 1], + strict=True, + ): + assert tool == { + "name": name, + "call_id": call_id, + "chunk_index": index, + "argument_chars": len(arguments), + "argument_bytes": len(arguments.encode()), + "argument_sha256": hashlib.sha256(arguments.encode()).hexdigest(), + "argument_deltas": count, + "argument_prefixes": [], + "argument_encoding_valid": True, + } assert attempt.data["response_status"] == "completed" assert "private" not in json.dumps(engine.records, default=str) assert "secret" not in json.dumps(engine.records, default=str) @@ -205,7 +275,10 @@ async def test_timeout_during_tool_generation_keeps_last_output_and_tool_name(mo ) assert failed["failure_phase"] == "generating_tool_arguments" assert failed["last_output_at"] >= failed["first_output_at"] - assert failed["tools"] == [{"name": "click", "argument_chars": 5}] + assert len(failed["tools"]) == 1 + tool = failed["tools"][0] + assert tool["name"] == "click" and tool["argument_chars"] == tool["argument_bytes"] == 5 + assert tool["argument_sha256"] == hashlib.sha256(b'{"x":').hexdigest() assert failed["usage"] is None assert failed["max_output_tokens"] == 12000