diff --git a/tests/integration/observability/test_grayswan_wire.py b/tests/integration/observability/test_grayswan_wire.py index 9c5a2d9be68..fc72ce5c27a 100644 --- a/tests/integration/observability/test_grayswan_wire.py +++ b/tests/integration/observability/test_grayswan_wire.py @@ -64,6 +64,12 @@ def _grayswan_config( *, on_flagged_action: str = "monitor", streaming_end_of_stream_only: bool = False, + default_on: bool = True, + fail_open: bool | None = None, + skip_system: bool | None = None, + skip_tool: bool | None = None, + scan_only_tool_results: bool | None = None, + extra_guardrails: tuple[dict[str, JsonValue], ...] = (), ) -> Path: config: Final = { **yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()), @@ -73,17 +79,24 @@ def _grayswan_config( "litellm_params": { "guardrail": "grayswan", "mode": mode, - "default_on": True, + "default_on": default_on, "api_base": vendor_url, "api_key": _VENDOR_KEY, "streaming_end_of_stream_only": streaming_end_of_stream_only, + **({"skip_system_message_in_guardrail": skip_system} if skip_system is not None else {}), + **({"skip_tool_message_in_guardrail": skip_tool} if skip_tool is not None else {}), + **( + {"scan_only_tool_results": scan_only_tool_results} if scan_only_tool_results is not None else {} + ), "optional_params": { "on_flagged_action": on_flagged_action, "violation_threshold": 0.5, "policy_id": "synthetic-policy", + **({"fail_open": fail_open} if fail_open is not None else {}), }, }, - } + }, + *extra_guardrails, ], } path: Final = tmp_path / f"{identity}.yaml" @@ -103,6 +116,8 @@ def _vendor(violation: float = 0.0): def _chat_provider(message: dict[str, JsonValue]): def respond(request: Request) -> Reply: + if request.target == "/v1/models": + return Reply(body=b'{"data":[]}') assert request.target == "/chat/completions", request.target return Reply( body=json.dumps( @@ -120,7 +135,7 @@ def _chat_provider(message: dict[str, JsonValue]): return respond -def _monitor_bodies(vendor: Wire, expected: int = 1) -> tuple[dict[str, JsonValue], ...]: +def _monitor_bodies(vendor: Wire, expected: int = 1, seconds: float = 30) -> tuple[dict[str, JsonValue], ...]: collected: tuple[dict[str, JsonValue], ...] = () def drain_new() -> tuple[dict[str, JsonValue], ...]: @@ -135,7 +150,7 @@ def _monitor_bodies(vendor: Wire, expected: int = 1) -> tuple[dict[str, JsonValu ) return collected - return eventually(drain_new, lambda bodies: len(bodies) >= expected, seconds=30) + return eventually(drain_new, lambda bodies: len(bodies) >= expected, seconds=seconds) def test_post_call_sends_request_conversation_and_tools(gateway: Gateway, tmp_path: Path) -> None: @@ -214,6 +229,8 @@ def test_post_call_sends_anthropic_messages_conversation(gateway: Gateway, tmp_p response_text: Final = "inbox checked" def provider(request: Request) -> Reply: + if request.target == "/v1/models": + return Reply(body=b'{"data":[]}') assert request.target == "/v1/messages", request.target return Reply( body=json.dumps( @@ -295,6 +312,8 @@ def test_post_call_sends_responses_api_input(gateway: Gateway, tmp_path: Path) - response_text: Final = "thread summarized" def provider(request: Request) -> Reply: + if request.target == "/v1/models": + return Reply(body=b'{"data":[]}') assert request.target == "/responses", request.target return Reply( body=json.dumps( @@ -354,6 +373,8 @@ def test_post_call_streams_end_of_stream_with_conversation(gateway: Gateway, tmp response_text: Final = "streamed summary" def provider(request: Request) -> Reply: + if request.target == "/v1/models": + return Reply(body=b'{"data":[]}') assert request.target == "/chat/completions", request.target assert json.loads(request.body)["stream"] is True frames: Final = ( @@ -429,3 +450,1081 @@ def test_pre_call_payload_shape_unchanged(gateway: Gateway, tmp_path: Path) -> N {"role": "user", "content": user_text}, ], body assert "tools" not in body, body + + +def test_post_call_merges_text_and_tool_calls_into_one_message(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + response_text: Final = "Sending that email now." + tool_call: Final = { + "id": "call_send", + "type": "function", + "function": {"name": "send_email", "arguments": '{"to": "cfo@example.com", "body": "done"}'}, + } + + with ( + wire_server(_vendor()) as vendor, + wire_server( + _chat_provider({"role": "assistant", "content": response_text, "tool_calls": [tool_call]}) + ) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": [dict(tool) for tool in _TOOLS], + }, + ) + assert response.status_code == 200, response.text + (body,) = _monitor_bodies(vendor) + assert body["messages"] == [ + *[dict(message) for message in _REQUEST_MESSAGES], + {"role": "assistant", "content": response_text, "tool_calls": [tool_call]}, + ], body + assert body["tools"] == [dict(tool) for tool in _TOOLS], body + + +def test_post_call_multi_choice_texts_and_tool_calls_stay_split(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + tool_call: Final = { + "id": "call_send", + "type": "function", + "function": {"name": "send_email", "arguments": '{"to": "cfo@example.com", "body": "done"}'}, + } + + def provider(request: Request) -> Reply: + if request.target == "/v1/models": + return Reply(body=b'{"data":[]}') + assert request.target == "/chat/completions", request.target + return Reply( + body=json.dumps( + { + "id": "chatcmpl-grayswan", + "object": "chat.completion", + "created": 1700000000, + "model": "gpt-4o-mini", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "first answer", "tool_calls": [tool_call]}, + "finish_reason": "tool_calls", + }, + { + "index": 1, + "message": {"role": "assistant", "content": "second answer"}, + "finish_reason": "stop", + }, + ], + "usage": {"prompt_tokens": 5, "completion_tokens": 6, "total_tokens": 11}, + } + ).encode() + ) + + with wire_server(_vendor()) as vendor, wire_server(provider) as upstream: + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "n": 2, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": [dict(tool) for tool in _TOOLS], + }, + ) + assert response.status_code == 200, response.text + (body,) = _monitor_bodies(vendor) + assert body["messages"] == [ + *[dict(message) for message in _REQUEST_MESSAGES], + {"role": "assistant", "content": "first answer"}, + {"role": "assistant", "content": "second answer"}, + {"role": "assistant", "tool_calls": [tool_call]}, + ], body + + +def _chat_stream_provider(chunks: int): + def respond(request: Request) -> Reply: + if request.target == "/v1/models": + return Reply(body=b'{"data":[]}') + assert request.target == "/chat/completions", request.target + frames: Final = tuple( + f'data: {{"id":"chatcmpl-s","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{{"index":0,"delta":{{"content":"part{i} "}}}}]}}\n\n'.encode() + for i in range(chunks) + ) + return Reply( + content_type="text/event-stream", + chunks=( + b'data: {"id":"chatcmpl-s","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"role":"assistant","content":""}}]}\n\n', + *frames, + b'data: {"id":"chatcmpl-s","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n', + b"data: [DONE]\n\n", + ), + ) + + return respond + + +def test_post_call_sampled_stream_calls_each_carry_context(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + + with wire_server(_vendor()) as vendor, wire_server(_chat_stream_provider(12)) as upstream: + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 64, + "stream": True, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": [dict(tool) for tool in _TOOLS], + }, + ) + assert response.status_code == 200, response.text + bodies: Final = _monitor_bodies(vendor, expected=2) + assert len(bodies) >= 2, bodies + for body in bodies: + messages: Final = body["messages"] + assert isinstance(messages, list), body + assert messages[:-1] == [dict(message) for message in _REQUEST_MESSAGES], body + last: Final = messages[-1] + assert isinstance(last, dict) and last["role"] == "assistant" and last["content"], body + assert body["tools"] == [dict(tool) for tool in _TOOLS], body + + +def test_post_call_anthropic_stream_sends_conversation(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + user_text: Final = f"check my inbox {identity}" + response_text: Final = "streamed inbox checked" + + def provider(request: Request) -> Reply: + if request.target == "/v1/models": + return Reply(body=b'{"data":[]}') + assert request.target == "/v1/messages", request.target + frames: Final = ( + b'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_s","type":"message","role":"assistant","model":"claude-opus-5-5","content":[],"stop_reason":null,"usage":{"input_tokens":10,"output_tokens":1}}}\n\n', + b'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}\n\n', + b'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"streamed inbox"}}\n\n', + b'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" checked"}}\n\n', + b'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}\n\n', + b'event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":3}}\n\n', + b'event: message_stop\ndata: {"type":"message_stop"}\n\n', + ) + return Reply(content_type="text/event-stream", chunks=frames) + + with wire_server(_vendor()) as vendor, wire_server(provider) as upstream: + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model( + model=f"anthropic/{_LATEST_CLAUDE}", api_base=upstream.url, api_key=_PROVIDER_KEY + ) + response: Final = candidate.request( + "POST", + "/v1/messages", + { + "model": model, + "max_tokens": 16, + "stream": True, + "messages": [ + {"role": "user", "content": user_text}, + { + "role": "assistant", + "content": [{"type": "tool_use", "id": "toolu_inbox", "name": "read_inbox", "input": {}}], + }, + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_inbox", + "content": f"Inbox: {_INJECTED}", + } + ], + }, + ], + }, + ) + assert response.status_code == 200, response.text + bodies: Final = _monitor_bodies(vendor, expected=1) + body: Final = bodies[-1] + messages: Final = body["messages"] + assert isinstance(messages, list), body + assert any( + isinstance(message, dict) + and message.get("role") == "user" + and user_text in str(message.get("content", "")) + for message in messages + ), body + last: Final = messages[-1] + assert isinstance(last, dict) and last["role"] == "assistant", body + assert response_text in str(last.get("content", "")), body + + +def test_post_call_responses_stream_sends_conversation(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + input_text: Final = f"summarize this thread {identity}" + response_text: Final = "streamed thread" + + def provider(request: Request) -> Reply: + if request.target == "/v1/models": + return Reply(body=b'{"data":[]}') + assert request.target == "/responses", request.target + output_item: Final = { + "type": "message", + "id": "msg_s", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": response_text, "annotations": []}], + } + frames: Final = ( + b'data: {"type":"response.created","response":{"id":"resp_s","object":"response","created_at":1700000000,"status":"in_progress","model":"gpt-5.3-codex","output":[]}}\n\n', + b'data: {"type":"response.output_item.added","output_index":0,"item":{"type":"message","id":"msg_s","status":"in_progress","role":"assistant","content":[]}}\n\n', + b'data: {"type":"response.output_text.delta","item_id":"msg_s","output_index":0,"content_index":0,"delta":"streamed "}\n\n', + b'data: {"type":"response.output_text.delta","item_id":"msg_s","output_index":0,"content_index":0,"delta":"thread"}\n\n', + f'data: {{"type":"response.output_item.done","output_index":0,"item":{json.dumps(output_item)}}}\n\n'.encode(), + f'data: {{"type":"response.completed","response":{{"id":"resp_s","object":"response","created_at":1700000000,"status":"completed","model":"gpt-5.3-codex","output":[{json.dumps(output_item)}],"usage":{{"input_tokens":5,"output_tokens":3,"total_tokens":8}}}}}}\n\n'.encode(), + ) + return Reply(content_type="text/event-stream", chunks=frames) + + responses_tool: Final = { + "type": "function", + "name": "send_email", + "description": "Send an email", + "parameters": {"type": "object", "properties": {"to": {"type": "string"}}, "required": ["to"]}, + } + with wire_server(_vendor()) as vendor, wire_server(provider) as upstream: + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model( + model="openai/responses/gpt-5.3-codex", api_base=upstream.url, api_key=_PROVIDER_KEY + ) + response: Final = candidate.request( + "POST", + "/v1/responses", + { + "model": model, + "stream": True, + "instructions": "You are terse.", + "input": [{"role": "user", "content": input_text}], + "tools": [responses_tool], + }, + ) + assert response.status_code == 200, response.text + bodies: Final = _monitor_bodies(vendor, expected=1) + body: Final = bodies[-1] + messages: Final = body["messages"] + assert isinstance(messages, list), body + assert any( + isinstance(message, dict) + and message.get("role") == "user" + and input_text in json.dumps(message.get("content", "")) + for message in messages + ), body + assert any( + isinstance(message, dict) + and message.get("role") == "assistant" + and response_text in str(message.get("content", "")) + for message in messages + ), body + assert body.get("tools") == [responses_tool], body + + +def test_post_call_openai_sdk_sync_and_async(gateway: Gateway, tmp_path: Path) -> None: + import asyncio + + import openai + + identity: Final = "grayswan" + uuid.uuid4().hex + response_text: Final = "sdk control" + + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + base_url: Final = str(candidate.client.base_url).rstrip("/") + request_body: Final = { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": [dict(tool) for tool in _TOOLS], + } + sync_client: Final = openai.OpenAI(base_url=f"{base_url}/v1", api_key=candidate.key) + sync_response: Final = sync_client.chat.completions.create(**request_body) + assert sync_response.choices[0].message.content == response_text + async_client: Final = openai.AsyncOpenAI(base_url=f"{base_url}/v1", api_key=candidate.key) + + async def call() -> str | None: + completed: Final = await async_client.chat.completions.create(**request_body) + return completed.choices[0].message.content + + assert asyncio.run(call()) == response_text + bodies: Final = _monitor_bodies(vendor, expected=2) + for body in bodies: + assert body["messages"] == [ + *[dict(message) for message in _REQUEST_MESSAGES], + {"role": "assistant", "content": response_text}, + ], body + assert body["tools"] == [dict(tool) for tool in _TOOLS], body + + +def test_post_call_anthropic_sdk_sends_conversation(gateway: Gateway, tmp_path: Path) -> None: + import anthropic + + identity: Final = "grayswan" + uuid.uuid4().hex + user_text: Final = f"check my inbox {identity}" + response_text: Final = "sdk inbox checked" + + def provider(request: Request) -> Reply: + if request.target == "/v1/models": + return Reply(body=b'{"data":[]}') + assert request.target == "/v1/messages", request.target + return Reply( + body=json.dumps( + { + "id": "msg_synthetic", + "type": "message", + "role": "assistant", + "model": _LATEST_CLAUDE, + "content": [{"type": "text", "text": response_text}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 10, "output_tokens": 3}, + } + ).encode() + ) + + with wire_server(_vendor()) as vendor, wire_server(provider) as upstream: + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model( + model=f"anthropic/{_LATEST_CLAUDE}", api_base=upstream.url, api_key=_PROVIDER_KEY + ) + client: Final = anthropic.Anthropic(base_url=str(candidate.client.base_url), api_key=candidate.key) + reply: Final = client.messages.create( + model=model, + max_tokens=16, + messages=[ + {"role": "user", "content": user_text}, + { + "role": "assistant", + "content": [{"type": "tool_use", "id": "toolu_inbox", "name": "read_inbox", "input": {}}], + }, + { + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "toolu_inbox", "content": f"Inbox: {_INJECTED}"} + ], + }, + ], + ) + assert response_text in reply.content[0].text + (body,) = _monitor_bodies(vendor) + messages: Final = body["messages"] + assert isinstance(messages, list), body + assert any( + isinstance(message, dict) + and message.get("role") == "user" + and user_text in str(message.get("content", "")) + for message in messages + ), body + last: Final = messages[-1] + assert isinstance(last, dict) and last["role"] == "assistant" and last["content"] == response_text, body + + +def _run_context_request( + gateway: Gateway, + tmp_path: Path, + *, + messages: list[dict[str, JsonValue]], + tools: list[dict[str, JsonValue]] | None, + expected_messages: list[dict[str, JsonValue]], + expect_tools: bool, + **config_kwargs: JsonValue, +) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + response_text: Final = "context control" + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call", **config_kwargs) + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "messages": messages, + **({"tools": tools} if tools is not None else {}), + }, + ) + assert response.status_code == 200, response.text + (body,) = _monitor_bodies(vendor) + assert body["messages"] == expected_messages, body + if expect_tools: + assert body["tools"] == tools, body + else: + assert "tools" not in body, body + + +def test_post_call_skip_system_message_drops_system_from_context(gateway: Gateway, tmp_path: Path) -> None: + request_messages: Final = [dict(message) for message in _REQUEST_MESSAGES] + _run_context_request( + gateway, + tmp_path, + messages=request_messages, + tools=[dict(tool) for tool in _TOOLS], + expected_messages=[ + *[dict(message) for message in _REQUEST_MESSAGES[1:]], + {"role": "assistant", "content": "context control"}, + ], + expect_tools=True, + skip_system=True, + ) + + +def test_post_call_skip_tool_message_drops_tool_from_context(gateway: Gateway, tmp_path: Path) -> None: + request_messages: Final = [dict(message) for message in _REQUEST_MESSAGES] + _run_context_request( + gateway, + tmp_path, + messages=request_messages, + tools=[dict(tool) for tool in _TOOLS], + expected_messages=[ + *[dict(message) for message in _REQUEST_MESSAGES[:3]], + {"role": "assistant", "content": "context control"}, + ], + expect_tools=True, + skip_tool=True, + ) + + +def test_post_call_scan_only_tool_results_scopes_context(gateway: Gateway, tmp_path: Path) -> None: + request_messages: Final = [dict(message) for message in _REQUEST_MESSAGES] + _run_context_request( + gateway, + tmp_path, + messages=request_messages, + tools=[dict(tool) for tool in _TOOLS], + expected_messages=[ + dict(_REQUEST_MESSAGES[3]), + {"role": "assistant", "content": "context control"}, + ], + expect_tools=False, + scan_only_tool_results=True, + ) + + +def test_post_call_all_messages_scoped_out_sends_response_only(gateway: Gateway, tmp_path: Path) -> None: + _run_context_request( + gateway, + tmp_path, + messages=[{"role": "system", "content": "only a system prompt"}], + tools=[dict(tool) for tool in _TOOLS], + expected_messages=[{"role": "assistant", "content": "context control"}], + expect_tools=False, + skip_system=True, + ) + + +def test_post_call_skip_flags_explicit_false_matches_default(gateway: Gateway, tmp_path: Path) -> None: + request_messages: Final = [dict(message) for message in _REQUEST_MESSAGES] + _run_context_request( + gateway, + tmp_path, + messages=request_messages, + tools=[dict(tool) for tool in _TOOLS], + expected_messages=[ + *request_messages, + {"role": "assistant", "content": "context control"}, + ], + expect_tools=True, + skip_system=False, + skip_tool=False, + scan_only_tool_results=False, + ) + + +def test_post_call_monitor_mode_flag_on_tool_call_only_response(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + tool_call: Final = { + "id": "call_send_email", + "type": "function", + "function": {"name": "send_email", "arguments": '{"to": "cfo@example.com", "body": "wire funds"}'}, + } + + with ( + wire_server(_vendor(violation=1.0)) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": None, "tool_calls": [tool_call]})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call", on_flagged_action="monitor") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": [dict(tool) for tool in _TOOLS], + }, + ) + assert response.status_code == 200, response.text + (body,) = _monitor_bodies(vendor) + messages: Final = body["messages"] + assert isinstance(messages, list), body + assert messages[:-1] == [dict(message) for message in _REQUEST_MESSAGES], body + last: Final = messages[-1] + assert isinstance(last, dict) and last["role"] == "assistant" and last.get("tool_calls"), body + + +def test_post_call_guardrail_attached_per_request_and_per_key(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + response_text: Final = "attached control" + + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call", default_on=False) + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + body_template: Final = { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": [dict(tool) for tool in _TOOLS], + } + per_request: Final = candidate.request( + "POST", "/v1/chat/completions", {**body_template, "guardrails": [identity]} + ) + assert per_request.status_code == 200, per_request.text + scoped_key: Final = scenario.key(metadata={"guardrails": [identity]}) + per_key: Final = candidate.request("POST", "/v1/chat/completions", body_template, key=scoped_key) + assert per_key.status_code == 200, per_key.text + bodies: Final = _monitor_bodies(vendor, expected=2) + for body in bodies: + assert body["messages"] == [ + *[dict(message) for message in _REQUEST_MESSAGES], + {"role": "assistant", "content": response_text}, + ], body + + +def test_post_call_cache_hit_still_sends_context(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + response_text: Final = "cached control" + + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + request_body: Final = { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": [dict(tool) for tool in _TOOLS], + } + first: Final = candidate.request("POST", "/v1/chat/completions", request_body) + assert first.status_code == 200, first.text + second: Final = candidate.request("POST", "/v1/chat/completions", request_body) + assert second.status_code == 200, second.text + bodies: Final = _monitor_bodies(vendor, expected=2) + for body in bodies: + assert body["messages"] == [ + *[dict(message) for message in _REQUEST_MESSAGES], + {"role": "assistant", "content": response_text}, + ], body + assert len(upstream.drain()) == 1 + + +def test_post_call_text_completion_surface_sends_response_only(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + response_text: Final = "completion done" + + def provider(request: Request) -> Reply: + assert request.target == "/completions", request.target + return Reply( + body=json.dumps( + { + "id": "cmpl-synthetic", + "object": "text_completion", + "created": 1700000000, + "model": "gpt-3.5-turbo-instruct", + "choices": [{"text": response_text, "index": 0, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 4, "completion_tokens": 2, "total_tokens": 6}, + } + ).encode() + ) + + with wire_server(_vendor()) as vendor, wire_server(provider) as upstream: + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model( + model="openai/gpt-3.5-turbo-instruct", api_base=upstream.url, api_key=_PROVIDER_KEY + ) + response: Final = candidate.request( + "POST", + "/v1/completions", + {"model": model, "prompt": "finish this sentence", "max_tokens": 4}, + ) + assert response.status_code == 200, response.text + (body,) = _monitor_bodies(vendor) + assert body["messages"] == [{"role": "assistant", "content": response_text}], body + assert "tools" not in body, body + + +def test_post_call_generic_guardrail_inputs_unchanged(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + generic_name: Final = "generic" + uuid.uuid4().hex + response_text: Final = "family control" + + def generic_policy(request: Request) -> Reply: + assert request.target == "/beta/litellm_basic_guardrail_api", request.target + return Reply(body=json.dumps({"action": "NONE"}).encode()) + + generic_entry: Final = { + "guardrail_name": generic_name, + "litellm_params": { + "guardrail": "generic_guardrail_api", + "mode": "post_call", + "default_on": True, + "api_base": None, + "api_key": "synthetic-guardrail-key", + }, + } + + with ( + wire_server(_vendor()) as vendor, + wire_server(generic_policy) as policy, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + generic_entry["litellm_params"]["api_base"] = ( + policy.url + ) # writable-ok: wire port only exists inside the context + config_path: Final = _grayswan_config( + tmp_path, identity, vendor.url, "post_call", extra_guardrails=(generic_entry,) + ) + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": [dict(tool) for tool in _TOOLS], + }, + ) + assert response.status_code == 200, response.text + (grayswan_body,) = _monitor_bodies(vendor) + generic_bodies: Final = eventually( + lambda: tuple( + _JSON_OBJECT.validate_json(request.body) + for request in policy.drain() + if request.target == "/beta/litellm_basic_guardrail_api" + ), + lambda bodies: len(bodies) >= 1, + seconds=30, + ) + generic_body: Final = generic_bodies[0] + assert generic_body["texts"] == [response_text], generic_body + assert grayswan_body["messages"][:-1] == [dict(message) for message in _REQUEST_MESSAGES], grayswan_body + + +def test_post_call_tools_in_invalid_shapes_omit_tools_key(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + response_text: Final = "no tools forwarded" + request_tools: Final = [dict(tool) for tool in _TOOLS] + + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + statuses: Final = tuple( + candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": tools_value, + }, + ).status_code + for tools_value in (request_tools[0], "send_email") + ) + assert all(status < 500 for status in statuses), statuses + expected_bodies: Final = sum(1 for status in statuses if status == 200) + bodies: Final = _monitor_bodies(vendor, expected=expected_bodies) if expected_bodies else vendor.drain() + for request in bodies: + body: Final = request if isinstance(request, dict) else _JSON_OBJECT.validate_json(request.body) + assert "tools" not in body, body + + +def test_post_call_user_content_parts_carried_verbatim(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + parts: Final = [ + {"type": "text", "text": "first part"}, + {"type": "text", "text": "second part"}, + ] + request_messages: Final = [ + dict(_REQUEST_MESSAGES[0]), + {"role": "user", "content": parts}, + *[dict(message) for message in _REQUEST_MESSAGES[2:]], + ] + response_text: Final = "parts control" + + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + {"model": model, "max_tokens": 16, "messages": request_messages}, + ) + assert response.status_code == 200, response.text + (body,) = _monitor_bodies(vendor) + messages: Final = body["messages"] + assert isinstance(messages, list), body + user_part_messages: Final = [ + message for message in messages if isinstance(message, dict) and message.get("role") == "user" + ] + assert any( + isinstance(message.get("content"), list) + and any(isinstance(part, dict) and part.get("text") == "second part" for part in message["content"]) + for message in user_part_messages + ), body + + +def test_post_call_large_and_repeated_messages_carried_verbatim(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + big_text: Final = "payload-" + "x" * 5000 + request_messages: Final = [ + dict(_REQUEST_MESSAGES[0]), + {"role": "user", "content": big_text}, + dict(_REQUEST_MESSAGES[2]), + dict(_REQUEST_MESSAGES[3]), + {"role": "user", "content": big_text}, + ] + response_text: Final = "big control" + + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + {"model": model, "max_tokens": 16, "messages": request_messages}, + ) + assert response.status_code == 200, response.text + (body,) = _monitor_bodies(vendor) + messages: Final = body["messages"] + assert isinstance(messages, list), body + big_copies: Final = [ + message + for message in messages + if isinstance(message, dict) and message.get("role") == "user" and message.get("content") == big_text + ] + assert len(big_copies) == 2, body + + +def test_post_call_vendor_500_fail_open_and_fail_closed(gateway: Gateway, tmp_path: Path) -> None: + response_text: Final = "vendor error control" + + def vendor_500(request: Request) -> Reply: + return Reply(status=500, body=b'{"error":"vendor down"}') + + def attempt(fail_open: bool, request_mark: str) -> int: + identity: Final = f"grayswan{request_mark}" + with ( + wire_server(vendor_500) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call", fail_open=fail_open) + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + messages_for_attempt: Final = [ + *_REQUEST_MESSAGES[:1], + {**_REQUEST_MESSAGES[1], "content": f"summarize my inbox {request_mark}"}, + *_REQUEST_MESSAGES[2:], + ] + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in messages_for_attempt], + }, + ) + assert len(upstream.drain()) == 1 + return response.status_code + + assert attempt(True, uuid.uuid4().hex) == 200 + assert attempt(False, uuid.uuid4().hex) >= 400 + + +def test_post_call_vendor_403_and_404_fail_open_and_fail_closed(gateway: Gateway, tmp_path: Path) -> None: + import itertools + + response_text: Final = "vendor auth error control" + statuses: Final = itertools.cycle((403, 404)) + + def vendor_respond(request: Request) -> Reply: + assert request.target == "/cygnal/monitor", request.target + return Reply(status=next(statuses), body=b'{"error":"vendor rejected"}') + + identity: Final = "grayswan" + uuid.uuid4().hex + with ( + wire_server(vendor_respond) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call", fail_open=True) + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + for index in range(2): + messages_for_attempt: Final = [ + *_REQUEST_MESSAGES[:1], + {**_REQUEST_MESSAGES[1], "content": f"summarize my inbox {index}"}, + *_REQUEST_MESSAGES[2:], + ] + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in messages_for_attempt], + }, + ) + assert response.status_code == 200, response.text + assert len(upstream.drain()) == 2 + + statuses2: Final = itertools.cycle((403, 404)) + + def vendor_respond_fresh(request: Request) -> Reply: + return Reply(status=next(statuses2), body=b'{"error":"vendor rejected"}') + + identity2: Final = "grayswan" + uuid.uuid4().hex + with ( + wire_server(vendor_respond_fresh) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path2: Final = _grayswan_config(tmp_path, identity2, vendor.url, "post_call", fail_open=False) + with owned_proxy(gateway, tmp_path, {}, config=config_path2) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + for index in range(2): + messages_for_attempt: Final = [ + *_REQUEST_MESSAGES[:1], + {**_REQUEST_MESSAGES[1], "content": f"summarize my inbox closed {index}"}, + *_REQUEST_MESSAGES[2:], + ] + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in messages_for_attempt], + }, + ) + assert response.status_code >= 400, response.text + assert len(upstream.drain()) == 2 + + +def test_post_call_assistant_tool_call_missing_id_no_500(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + request_messages: Final = [ + dict(_REQUEST_MESSAGES[0]), + dict(_REQUEST_MESSAGES[1]), + { + "role": "assistant", + "tool_calls": [{"type": "function", "function": {"name": "read_inbox", "arguments": "{}"}}], + }, + dict(_REQUEST_MESSAGES[3]), + ] + response_text: Final = "missing id control" + + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + {"model": model, "max_tokens": 16, "messages": request_messages}, + ) + assert response.status_code < 500, response.text + (body,) = _monitor_bodies(vendor) + messages: Final = body["messages"] + assert isinstance(messages, list) and messages, body + + +def test_post_call_responses_string_input_becomes_user_message(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + input_text: Final = f"plain string input {identity}" + response_text: Final = "string input done" + + def provider(request: Request) -> Reply: + if request.target == "/v1/models": + return Reply(body=b'{"data":[]}') + assert request.target == "/responses", request.target + return Reply( + body=json.dumps( + { + "id": "resp_synthetic", + "object": "response", + "created_at": 1700000000, + "status": "completed", + "model": "gpt-5.3-codex", + "output": [ + { + "type": "message", + "id": "msg_synthetic", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": response_text, "annotations": []}], + } + ], + "usage": {"input_tokens": 5, "output_tokens": 3, "total_tokens": 8}, + } + ).encode() + ) + + with wire_server(_vendor()) as vendor, wire_server(provider) as upstream: + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model( + model="openai/responses/gpt-5.3-codex", api_base=upstream.url, api_key=_PROVIDER_KEY + ) + response: Final = candidate.request("POST", "/v1/responses", {"model": model, "input": input_text}) + assert response.status_code == 200, response.text + (body,) = _monitor_bodies(vendor) + messages: Final = body["messages"] + assert isinstance(messages, list), body + assert any( + isinstance(message, dict) + and message.get("role") == "user" + and input_text in str(message.get("content", "")) + for message in messages + ), body + last: Final = messages[-1] + assert isinstance(last, dict) and last["role"] == "assistant" and last["content"] == response_text, body + + +def test_post_call_empty_and_missing_tools_omit_tools_key(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + response_text: Final = "empty tools control" + + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + for tools_value in ([], None): + request_body: Final = { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + **({"tools": tools_value} if tools_value is not None else {}), + } + response: Final = candidate.request("POST", "/v1/chat/completions", request_body) + assert response.status_code == 200, response.text + bodies: Final = _monitor_bodies(vendor, expected=2) + assert len(bodies) == 2, bodies + for body in bodies: + assert "tools" not in body, body + assert body["messages"][:-1] == [dict(message) for message in _REQUEST_MESSAGES], body + + +def test_post_call_five_identical_requests_each_send_context(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + response_text: Final = "idempotent control" + + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + request_body: Final = { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": [dict(tool) for tool in _TOOLS], + } + for _ in range(5): + response: Final = candidate.request("POST", "/v1/chat/completions", request_body) + assert response.status_code == 200, response.text + bodies: Final = _monitor_bodies(vendor, expected=5) + assert len(bodies) == 5, bodies + for body in bodies: + assert body["messages"] == [ + *[dict(message) for message in _REQUEST_MESSAGES], + {"role": "assistant", "content": response_text}, + ], body + + +def test_post_call_dynamic_extra_body_merged_with_context(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + response_text: Final = "dynamic params control" + + with ( + wire_server(_vendor()) as vendor, + wire_server(_chat_provider({"role": "assistant", "content": response_text})) as upstream, + ): + config_path: Final = _grayswan_config(tmp_path, identity, vendor.url, "post_call") + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate, candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "messages": [dict(message) for message in _REQUEST_MESSAGES], + "tools": [dict(tool) for tool in _TOOLS], + "guardrails": [{identity: {"extra_body": {"metadata": {"audit": "e5"}}}}], + }, + ) + assert response.status_code == 200, response.text + (body,) = _monitor_bodies(vendor) + assert body["messages"][:-1] == [dict(message) for message in _REQUEST_MESSAGES], body + assert body["tools"] == [dict(tool) for tool in _TOOLS], body + assert body.get("metadata") == {"audit": "e5"}, body diff --git a/tests/integration/observability/test_grayswan_wire_chaos.py b/tests/integration/observability/test_grayswan_wire_chaos.py new file mode 100644 index 00000000000..b78559e1d5b --- /dev/null +++ b/tests/integration/observability/test_grayswan_wire_chaos.py @@ -0,0 +1,223 @@ +import json +import os +import signal +import threading +import time +import uuid +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import Final + +import yaml +from integration._support.client import Gateway +from integration._support.process import group_members, owned_proxy_process +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue, TypeAdapter +from test_grayswan_wire import _PROVIDER_KEY, _REQUEST_MESSAGES, _VENDOR_KEY, _monitor_bodies + +_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue]) + + +def _chaos_config(tmp_path: Path, identity: str, vendor_url: str, *, fail_open: bool = True) -> Path: + config: Final = { + **yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()), + "guardrails": [ + { + "guardrail_name": identity, + "litellm_params": { + "guardrail": "grayswan", + "mode": "post_call", + "default_on": True, + "api_base": vendor_url, + "api_key": _VENDOR_KEY, + "streaming_end_of_stream_only": True, + "optional_params": { + "on_flagged_action": "monitor", + "violation_threshold": 0.5, + "policy_id": "synthetic-policy", + "fail_open": fail_open, + }, + }, + } + ], + } + path: Final = tmp_path / f"{identity}.yaml" + path.write_text(yaml.safe_dump(config)) + return path + + +def _provider(request: Request) -> Reply: + if request.method != "POST" or not request.body: + return Reply(body=b"{}") + body: Final = json.loads(request.body) + marker: Final = next( + ( + str(message.get("content")) + for message in body.get("messages", []) + if isinstance(message, dict) and str(message.get("content", "")).startswith("marker-") + ), + "none", + ) + if body.get("stream"): + frames: Final = ( + b'data: {"id":"chatcmpl-c","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"role":"assistant","content":""}}]}\n\n', + f'data: {{"id":"chatcmpl-c","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{{"index":0,"delta":{{"content":"echo {marker}"}}}}]}}\n\n'.encode(), + b'data: {"id":"chatcmpl-c","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n', + b"data: [DONE]\n\n", + ) + return Reply(content_type="text/event-stream", chunks=frames) + return Reply( + body=json.dumps( + { + "id": "chatcmpl-chaos", + "object": "chat.completion", + "created": 1700000000, + "model": "gpt-4o-mini", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": f"echo {marker}"}, + "finish_reason": "stop", + } + ], + "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8}, + } + ).encode() + ) + + +def _fire(candidate: Gateway, model: str, marker: str, stream: bool) -> int: + response: Final = candidate.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "max_tokens": 16, + "stream": stream, + "messages": [ + dict(_REQUEST_MESSAGES[0]), + {"role": "user", "content": marker}, + *[dict(message) for message in _REQUEST_MESSAGES[2:]], + ], + }, + ) + response.read() + return response.status_code + + +def _body_markers(body: dict[str, JsonValue]) -> tuple[str, ...]: + messages: Final = body.get("messages") + if not isinstance(messages, list): + return () + return tuple( + str(message.get("content")) + for message in messages + if isinstance(message, dict) + and isinstance(message.get("content"), str) + and message["content"].startswith("marker-") + ) + + +def test_vendor_outage_mid_burst_no_duplicate_monitor_calls(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + up: Final = threading.Event() + up.set() + + def vendor(request: Request) -> Reply: + assert request.target == "/cygnal/monitor", request.target + assert request.headers["grayswan-api-key"] == _VENDOR_KEY + if not up.is_set(): + return Reply(status=503, body=b'{"error":"sink down"}') + return Reply(body=b'{"violation":0.0}') + + with wire_server(vendor) as vendor_wire, wire_server(_provider) as upstream: + config_path: Final = _chaos_config(tmp_path, identity, vendor_wire.url) + with owned_proxy_process(gateway, tmp_path, {}, config=config_path, workers=2) as owned: + candidate: Final = owned.gateway + with candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + with ThreadPoolExecutor(max_workers=10) as pool: + before: Final = tuple( + pool.map(lambda i: _fire(candidate, model, f"marker-up-{i}", i < 2), range(8)) + ) + assert all(status == 200 for status in before), before + first_bodies: Final = _monitor_bodies(vendor_wire, expected=8) + up.clear() + during: Final = tuple( + pool.map(lambda i: _fire(candidate, model, f"marker-down-{i}", i < 2), range(8)) + ) + assert all(status == 200 for status in during), during + up.set() + after: Final = tuple( + pool.map(lambda i: _fire(candidate, model, f"marker-post-{i}", i < 2), range(8)) + ) + assert all(status == 200 for status in after), after + rest_bodies: Final = _monitor_bodies(vendor_wire, expected=16, seconds=50) + bodies: Final = (*first_bodies, *rest_bodies) + observed: Final = tuple(marker for body in bodies for marker in _body_markers(body)) + unique: Final = frozenset(observed) + assert len(observed) == len(unique), observed + for index in range(8): + assert f"marker-up-{index}" in unique, observed + assert f"marker-post-{index}" in unique, observed + for body in bodies: + messages: Final = body["messages"] + assert isinstance(messages, list) and len(messages) >= 2, body + assert any(isinstance(message, dict) and message.get("role") == "tool" for message in messages), ( + body + ) + + +def test_slow_vendor_burst_completes_without_deadlock(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + + def slow_vendor(request: Request) -> Reply: + assert request.target == "/cygnal/monitor", request.target + time.sleep(2) + return Reply(body=b'{"violation":0.0}') + + with wire_server(slow_vendor) as vendor, wire_server(_provider) as upstream: + config_path: Final = _chaos_config(tmp_path, identity, vendor.url) + with owned_proxy_process(gateway, tmp_path, {}, config=config_path, workers=2) as owned: + candidate: Final = owned.gateway + with candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + with ThreadPoolExecutor(max_workers=10) as pool: + statuses: Final = tuple( + pool.map(lambda i: _fire(candidate, model, f"marker-slow-{i}", False), range(10)) + ) + assert all(status == 200 for status in statuses), statuses + bodies: Final = _monitor_bodies(vendor, expected=10) + assert len(bodies) == 10, bodies + for body in bodies: + assert _body_markers(body), body + + +def test_worker_kill_mid_burst_survivor_keeps_serving(gateway: Gateway, tmp_path: Path) -> None: + identity: Final = "grayswan" + uuid.uuid4().hex + + def vendor(request: Request) -> Reply: + return Reply(body=b'{"violation":0.0}') + + with wire_server(vendor) as vendor_wire, wire_server(_provider) as upstream: + config_path: Final = _chaos_config(tmp_path, identity, vendor_wire.url) + with owned_proxy_process(gateway, tmp_path, {}, config=config_path, workers=2) as owned: + candidate: Final = owned.gateway + with candidate.scenario() as scenario: + model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=upstream.url, api_key=_PROVIDER_KEY) + warm: Final = _fire(candidate, model, "marker-warm", False) + assert warm == 200 + members: Final = group_members(owned.process.pid) + children: Final = tuple(member for member in members if member.pid != owned.process.pid) + assert len(children) >= 2, [member.pid for member in members] + os.kill(children[0].pid, signal.SIGKILL) + statuses: Final = tuple(_fire(candidate, model, f"marker-kill-{index}", False) for index in range(6)) + assert all(status == 200 for status in statuses), statuses + bodies: Final = _monitor_bodies(vendor_wire, expected=7) + kill_bodies: Final = [ + body for body in bodies if any(m.startswith("marker-kill-") for m in _body_markers(body)) + ] + assert len(kill_bodies) == 6, bodies + for body in kill_bodies: + messages: Final = body["messages"] + assert isinstance(messages, list) and len(messages) >= 2, body