test(integration): make rc/1.103.0 integration groups collect and pass again (#43382)

Adds the OwnedProxy harness the #42486 backport's OTel test imports, registers the #43029 backport's spend flush test in contracts.json with its covers marker, and aligns the MCP tool failure assertion with main.

tests/integration/_support/process.py

@@ -7,6 +7,7 @@ import time
 import uuid
 from collections.abc import Iterator, Mapping
 from contextlib import contextmanager
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Final
 
@@ -45,8 +46,37 @@ def stop_root_process(process: subprocess.Popen[bytes]) -> bool:
     return True
 
 
+@dataclass(frozen=True, slots=True)
+class OwnedProxy:
+    gateway: Gateway
+    process: subprocess.Popen[bytes]
+    log: Path
+
+
 @contextmanager
-def owned_proxy(gateway: Gateway, directory: Path, overrides: Mapping[str, str], *, config: Path | None = None, remove_environment: tuple[str, ...] = ()) -> Iterator[Gateway]:
+def owned_proxy(
+    gateway: Gateway,
+    directory: Path,
+    overrides: Mapping[str, str],
+    *,
+    config: Path | None = None,
+    remove_environment: tuple[str, ...] = (),
+) -> Iterator[Gateway]:
+    with owned_proxy_process(
+        gateway, directory, overrides, config=config, remove_environment=remove_environment
+    ) as owned:
+        yield owned.gateway
+
+
+@contextmanager
+def owned_proxy_process(
+    gateway: Gateway,
+    directory: Path,
+    overrides: Mapping[str, str],
+    *,
+    config: Path | None = None,
+    remove_environment: tuple[str, ...] = (),
+) -> Iterator[OwnedProxy]:
     with socket.socket() as reserve:
         reserve.bind(("127.0.0.1", 0))
         port: Final = reserve.getsockname()[1]
@@ -60,7 +90,8 @@ def owned_proxy(gateway: Gateway, directory: Path, overrides: Mapping[str, str],
     }
     output: Final = Path(os.environ.get("INTEGRATION_RESULTS_DIR", str(directory)))
     output.mkdir(parents=True, exist_ok=True)
-    with (output / f"owned-proxy-{uuid.uuid4().hex}.log").open("w") as log:
+    log_path: Final = output / f"owned-proxy-{uuid.uuid4().hex}.log"
+    with log_path.open("w") as log:
         process: Final = subprocess.Popen(
             [
                 sys.executable,
@@ -97,7 +128,7 @@ def owned_proxy(gateway: Gateway, directory: Path, overrides: Mapping[str, str],
                         pass
                     assert time.monotonic() < deadline, "Owned proxy readiness deadline exceeded"
                     time.sleep(0.1)
-                yield Gateway(client, gateway.key, gateway.upstream_url)
+                yield OwnedProxy(Gateway(client, gateway.key, gateway.upstream_url), process, log_path)
         finally:
             root_stopped: Final = stop_root_process(process)
             residual: Final = group_members(process.pid)

tests/integration/contracts.json

@@ -1380,6 +1380,9 @@
     ],
     "tests/integration/observability/test_otel_conversation_id.py::test_terminating_the_proxy_right_after_a_burst_flushes_every_span_before_exit": [
       "other.observability.otel.conversation_id_flushes_on_shutdown"
+    ],
+    "tests/integration/spend/test_team_member_spend_flush.py::test_fractional_member_spend_lands_after_a_whole_number_flush_on_the_same_connection": [
+      "quota_management.spend_tracking.team_member_spend_lands_after_zero_cost_flush"
     ]
   },
   "browser": {

tests/integration/mcp/test_mcp_lifecycle.py

@@ -53,7 +53,7 @@ def test_tool_error_remains_error_and_healthy_sibling_returns_value(gateway: Gat
         failure: Final = call_tool(gateway, key, identity, names["fail"], {})
         assert failure.status_code == 200, failure.text
         assert failure.json()["isError"] is True
-        assert "synthetic tool failure" in failure.json()["content"][0]["text"]
+        assert failure.json()["content"][0]["text"] == "Error executing tool fail"
         healthy: Final = call_tool(gateway, key, identity, names["multiply"], {"a": 3, "b": 5})
         assert healthy.status_code == 200, healthy.text
         assert healthy.json()["isError"] is False

tests/integration/spend/test_team_member_spend_flush.py

@@ -70,6 +70,7 @@ def _chat(gateway: Gateway, key: str, model: str) -> str:
     return string_value(response.json()["id"])
 
 
+@pytest.mark.covers("quota_management.spend_tracking.team_member_spend_lands_after_zero_cost_flush")
 def test_fractional_member_spend_lands_after_a_whole_number_flush_on_the_same_connection(
     gateway: Gateway, tmp_path: Path
 ) -> None: