From 45e947db53cea9ef024d6e2427ee173cd72f24b3 Mon Sep 17 00:00:00 2001 From: James Graham Date: Mon, 6 Jul 2026 20:17:07 +0100 Subject: [PATCH 1/2] Support testing with multiple versions of Firefox First write a test plan to check if this is likely a webcompat issue, and decide which platforms might be affected. Based on the results of that, run on some combination of nightly, stable and esr builds, and record the results for all of them. --- .../autowebcompat_repro/__main__.py | 12 +- .../autowebcompat_repro/agent.py | 364 +++++++++++++++--- .../autowebcompat_repro/browser.py | 74 ++++ .../autowebcompat_repro/firefox_install.py | 45 --- .../autowebcompat_repro/result.py | 93 +++-- 5 files changed, 467 insertions(+), 121 deletions(-) create mode 100644 agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/browser.py delete mode 100644 agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/firefox_install.py diff --git a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/__main__.py b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/__main__.py index c538f04480..4f08ed36b0 100644 --- a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/__main__.py +++ b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/__main__.py @@ -1,5 +1,6 @@ import logging from datetime import datetime +from typing import Literal from hackbot_runtime import ( HackbotAgentResult, @@ -26,7 +27,14 @@ class AgentInputs(BaseSettings): bug_id: int | None = None model: str | None = None max_turns: int | None = None - effort: str | None = None + effort: ( + Literal["low"] + | Literal["medium"] + | Literal["high"] + | Literal["xhigh"] + | Literal["max"] + | None + ) = None model_config = SettingsConfigDict(extra="ignore") @@ -42,7 +50,7 @@ async def main(ctx: HackbotContext) -> AutowebcompatResult: inputs = AgentInputs() # type: ignore if inputs.bug_data is not None: - input_data = BugDataInput(bug_data=inputs.bug_data) + input_data: BugDataInput | BugIdInput = BugDataInput(bug_data=inputs.bug_data) elif inputs.bug_id is not None: input_data = BugIdInput(bug_id=inputs.bug_id) diff --git a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/agent.py b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/agent.py index cc4a8a222d..529bbfe722 100644 --- a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/agent.py +++ b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/agent.py @@ -14,6 +14,7 @@ from collections.abc import Callable from dataclasses import dataclass from datetime import datetime +from enum import Enum from pathlib import Path from typing import Any, Generic, Literal @@ -27,16 +28,18 @@ from hackbot_runtime.claude import Reporter from pydantic import BaseModel +from .browser import FirefoxBrowsers from .config import BUGZILLA_READ_TOOLS, DEVTOOLS_TOOLS from .devtools_mcp import build_devtools_server -from .firefox_install import install_firefox_nightly from .result import ( RESULT_SERVER_NAME, SUBMIT_RESULT_TOOL, + BugReproductionResult, ChromeMaskResult, ReproductionResult, ResultCollector, ResultT, + TestPlanResult, build_result_server, ) from .setup_profile import setup_profile @@ -75,15 +78,24 @@ class AutowebcompatReproResult(BaseModel): summary: str failure_reason: str | None steps: str - screenshot_path: str | None - chrome_mask_fixed: bool | None = None + screenshot_url: str | None + plan_result: TestPlanResult + reproductions: list[tuple[str, BugReproductionResult | ReproductionResult]] + chrome_mask_fixed: bool | None @dataclass class TaskConfig: model: str | None = None max_turns: int | None = None - effort: str | None = None + effort: ( + Literal["low"] + | Literal["medium"] + | Literal["high"] + | Literal["xhigh"] + | Literal["max"] + | None + ) = None log: Path | None = None verbose: bool = True @@ -98,7 +110,7 @@ class TaskRun: class RunTracker: - def __init__(self): + def __init__(self) -> None: self.task_runs: list[TaskRun] = [] self.current_task: tuple[str, datetime] | None = None @@ -157,7 +169,9 @@ def __init__(self, task_config: TaskConfig, run_tracker: RunTracker): if result_server is not None: self.mcp_servers[self.result_server_name] = result_server - def add_mcp_server(self, name: str, server: McpServerConfig, tools: list[str]): + def add_mcp_server( + self, name: str, server: McpServerConfig, tools: list[str] + ) -> None: self.mcp_servers[name] = server self.allowed_tools.extend(tools) @@ -181,12 +195,12 @@ def agent_options(self) -> ClaudeAgentOptions: allowed_tools=self.allowed_tools, model=self.task_config.model, max_turns=self.task_config.max_turns, - **({"effort": self.task_config.effort} if self.task_config.effort else {}), setting_sources=[], # DevTools snapshots of complex pages serialize to JSON that can # exceed the SDK's default 1 MiB message buffer (the reader dies # fatally if it does). Raise it well above that ceiling. max_buffer_size=10 * 1024 * 1024, + effort=self.task_config.effort, ) async def run(self) -> ResultT: @@ -230,9 +244,69 @@ def make_empty_temp_file(dir: Path, prefix: str | None, suffix: str) -> Path: return Path(path) -class Reproduction(Task): - name = "reproduction" - result_cls = ReproductionResult +class TestPlan(Task): + name = "test_plan" + result_cls = TestPlanResult + + def __init__( + self, + task_config: TaskConfig, + run_tracker: RunTracker, + input_data: AutoWebcompatInput, + bugzilla_mcp_server: McpServerConfig, + ): + super().__init__(task_config, run_tracker) + self.input_data = input_data + if self.input_data.type == "bug_id": + self.add_mcp_server("bugzilla", bugzilla_mcp_server, BUGZILLA_READ_TOOLS) + + def subject(self) -> Any: + return self.input_data.subject() + + def system_prompt(self) -> str: + return ( + super() + .system_prompt() + .format( + task_details=""" +1. Identify the affected URL and the described broken behavior. + +2. If the report appears to describe something other than a webcompat issue i.e. +it doesn't meet the criteria under "Definition of a webcompat issue", +submit your findings via `submit_result` with `is_webcompat` set to `false`. + +3. Based on the report text determine which versions of Firefox are +likely to be affected by the issue. In particular: + - Is the issue described as affecting iOS? If so it is unlikely to affect other platforms. + - If the issue is not iOS, does it appear to affect desktop and Android, or only one or the other. + - If it affects desktop, is there evidence that the issue is specific to particular operating systems? + Note that often issues may only be reported on one operating system, but actually affect others. + An issue can only be assumed to be specific to a particular desktop operating system if it is stated + that it didn't reproduce on other platforms/ + - Is the issue marked as affecting nightly builds, stable builds, or ESR builds + +4. Submit your findings via `submit_result` (see "Reporting your result"). +""" + ) + ) + + def user_prompt(self) -> str: + if isinstance(self.input_data, BugDataInput): + return ( + "Here is the web-compatibility report to work on:\n\n" + f"{self.input_data.bug_data}\n\n" + "Follow your task procedure." + ) + if isinstance(self.input_data, BugIdInput): + return ( + f"The web-compatibility report to work on is Bugzilla bug {self.input_data.bug_id}. " + "Fetch it using the Bugzilla MCP tools, then follow your task procedure." + ) + + +class BugReproduction(Task): + name = "bug_reproduction" + result_cls = BugReproductionResult def __init__( self, @@ -301,6 +375,52 @@ def user_prompt(self) -> str: ) +class StepsReproduction(Task): + name = "steps_reproduction" + result_cls = ReproductionResult + + def __init__( + self, + task_config: TaskConfig, + run_tracker: RunTracker, + firefox_path: Path, + profile_path: Path, + steps: str, + ): + super().__init__(task_config, run_tracker) + self.steps = steps + self.add_mcp_server( + "firefox_devtools", + build_devtools_server( + firefox_path=firefox_path, + headless=True, + enable_script=True, + enable_privileged_context=False, + profile_path=profile_path, + ), + DEVTOOLS_TOOLS, + ) + + def subject(self) -> Any: + return self.steps + + def system_prompt(self) -> str: + return ( + super() + .system_prompt() + .format( + task_details=""" +1. Run the reproduction steps +2. Submit your findings via `submit_result` (see "Reporting your result"). +""" + ) + ) + + def user_prompt(self) -> str: + return f"""Here are the steps to reproduce the issue: +{self.steps}""" + + class ChromeMaskReproduction(Task): name = "chrome_mask" result_cls = ChromeMaskResult @@ -312,8 +432,9 @@ def __init__( firefox_path: Path, profile_path: Path, steps: str, - ): + ) -> None: super().__init__(task_config, run_tracker) + self.steps = steps self.add_mcp_server( "firefox_devtools", build_devtools_server( @@ -325,7 +446,6 @@ def __init__( ), DEVTOOLS_TOOLS, ) - self.steps = steps def subject(self) -> Any: return self.steps @@ -359,6 +479,108 @@ def user_prompt(self) -> str: {self.steps}""" +class FirefoxChannel(Enum): + nightly = "nightly" + stable = "stable" + esr = "esr" + + +@dataclass +class InitialReproduction: + channel: FirefoxChannel + steps: str + summary: str + screenshot_path: Path | None + + +class ReproductionResults: + def __init__(self, publish_file: PublishFile, plan_result: TestPlanResult): + self.plan_result = plan_result + self.publish_file = publish_file + self.results: dict[ + tuple[FirefoxChannel, str | None], ReproductionResult | ChromeMaskResult + ] = {} + self.initial_repro: InitialReproduction | None = None + self.chrome_mask_fixed: bool | None = None + + @property + def reproduced(self) -> bool: + return self.initial_repro is not None + + @property + def summary(self) -> str: + return self.initial_repro.summary if self.initial_repro is not None else "" + + @property + def failure_reason(self) -> str | None: + if self.reproduced: + return None + for result in self.results.values(): + # Return the first failure reason we got + if ( + isinstance(result, ReproductionResult) + and result.failure_reason is not None + ): + return result.failure_reason + return None + + @property + def steps(self) -> str: + return self.initial_repro.steps if self.initial_repro is not None else "" + + @property + def screenshot_url(self) -> str | None: + if ( + self.initial_repro is not None + and self.initial_repro.screenshot_path is not None + ): + return self.publish_file( + f"screenshot-{self.initial_repro.channel}.png", + self.initial_repro.screenshot_path, + "image/png", + ) + + return None + + def set_result( + self, + channel: FirefoxChannel, + extra: str | None, + result: ReproductionResult | ChromeMaskResult, + ) -> None: + key = (channel, extra) + if key in self.results: + raise ValueError(f"Got duplicate results for {channel}, {extra}") + if isinstance(result, BugReproductionResult) and result.reproduced: + if self.initial_repro is not None: + raise ValueError("Got duplicate steps / summary") + self.initial_repro = InitialReproduction( + channel, result.steps, result.summary, result.screenshot_path + ) + elif isinstance(result, ChromeMaskResult): + if self.chrome_mask_fixed is not None: + raise ValueError("Got duplicate results for chrome mask") + self.chrome_mask_fixed = result.chrome_mask_fixed + + self.results[key] = result + + def into_result(self) -> AutowebcompatReproResult: + return AutowebcompatReproResult( + reproduced=self.reproduced, + summary=self.summary, + failure_reason=self.failure_reason, + steps=self.steps, + screenshot_url=self.screenshot_url, + plan_result=self.plan_result, + reproductions=[ + (key[0].value, value.model_copy(update={"screenshot_path": None})) + for key, value in self.results.items() + if isinstance(value, ReproductionResult) + ], + chrome_mask_fixed=self.chrome_mask_fixed, + ) + + async def run_autowebcompat_repro( default_config: TaskConfig, tracker: RunTracker, @@ -371,48 +593,92 @@ async def run_autowebcompat_repro( Returns a :class:`AutowebcompatReproResult` on success; raises :class:`AgentError` if the agent ends in an error. """ - nightly_path = install_firefox_nightly() + firefox_browser = FirefoxBrowsers() + + test_plan_task = TestPlan(default_config, tracker, input_data, bugzilla_mcp_server) + test_plan_result = await test_plan_task.run() + repro_results = ReproductionResults(publish_file, test_plan_result) + + if not test_plan_result.is_webcompat: + result = repro_results.into_result() + result.summary = "Test was identified as a non-compat issue" + result.failure_reason = "non_compat" + return result + elif test_plan_result.affects_platforms == ["ios"]: + result = repro_results.into_result() + result.summary = "Issue was identified as iOS only" + result.failure_reason = "unsupported_platform" + return result + + async def next_repro_task( + channel: FirefoxChannel, + extra: str | None = None, + config: TaskConfig = default_config, + ) -> None: + browser = getattr(firefox_browser, channel.value) + profile = setup_profile(browser) + if repro_results.initial_repro is None: + task: Task = BugReproduction( + config, + tracker, + browser, + profile, + input_data, + bugzilla_mcp_server, + screenshots_dir, + ) + else: + task = StepsReproduction( + config, + tracker, + browser, + profile, + repro_results.initial_repro.steps, + ) + logger.info( + "Trying reproduction in %s%s", + channel, + f" {extra}" if extra is not None else "", + ) + repro_results.set_result(channel, extra, await task.run()) screenshots_dir = Path(tempfile.mkdtemp(prefix="autowebcompat-screenshots-")) # Always try in nightly first - repro_task = Reproduction( - default_config, - tracker, - nightly_path, - setup_profile(nightly_path, extensions=[]), - input_data, - bugzilla_mcp_server, - screenshots_dir, - ) - repro_result = await repro_task.run() - - if repro_result.screenshot_path is not None: - screenshot_path = publish_file( - "screenshot-nightly.png", repro_result.screenshot_path, "image/png" - ) - else: - screenshot_path = None - - result = AutowebcompatReproResult( - reproduced=repro_result.reproduced, - summary=repro_result.summary, - failure_reason=repro_result.failure_reason, - steps=repro_result.steps, - screenshot_path=screenshot_path, - ) - - if repro_result.reproduced: - # Build a profile with Chrome Mask preinstalled. - chrome_mask_profile = setup_profile(nightly_path, extensions=["chrome-mask"]) - chrome_mask_task = ChromeMaskReproduction( + await next_repro_task(FirefoxChannel.nightly) + + if not repro_results.reproduced and test_plan_result.affects_platforms == [ + "android" + ]: + result = repro_results.into_result() + result.summary = "Issue was identified as Android only and didn't reproduce on desktop nightly" + result.failure_reason = "unsupported_platform" + return result + + # If we don't think this is ESR only, try stable + if ( + "stable" in test_plan_result.affects_channels + or "nightly" in test_plan_result.affects_channels + ): + await next_repro_task(FirefoxChannel.stable) + + if repro_results.reproduced or "esr" in test_plan_result.affects_channels: + # If we have any result try ESR as a possible regression baseline, + # otherwise try ESR if we think it's affected + await next_repro_task(FirefoxChannel.esr) + + if repro_results.initial_repro is not None: + channel = repro_results.initial_repro.channel + browser = getattr(firefox_browser, channel.value) + profile = setup_profile(browser, extensions=["chrome-mask"]) + + task = ChromeMaskReproduction( default_config, tracker, - nightly_path, - chrome_mask_profile, - result.steps, + browser, + profile, + repro_results.initial_repro.steps, ) - chrome_mask_result = await chrome_mask_task.run() - result.chrome_mask_fixed = chrome_mask_result.chrome_mask_fixed + repro_results.set_result(channel, "chrome-mask", await task.run()) - return result + return repro_results.into_result() diff --git a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/browser.py b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/browser.py new file mode 100644 index 0000000000..189e3dd551 --- /dev/null +++ b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/browser.py @@ -0,0 +1,74 @@ +import logging +import platform +import tempfile +from pathlib import Path +from typing import Literal + +import mozdownload +import mozinstall + +logger = logging.getLogger("autowebcompat-repro") + + +def install_firefox( + channel: Literal["nightly"] | Literal["stable"] | Literal["esr"], +) -> Path: + install_dir = tempfile.mkdtemp(prefix=f"firefox-{channel}-", dir=Path.home()) + + # mozdownload doesn't correctly get arm builds for arm linux + mozdownload_platform = ( + "linux-arm64" if platform.machine() in ("aarch64", "arm64") else None + ) + + kwargs = {} + if channel == "nightly": + scraper_type = "daily" + kwargs["branch"] = "mozilla-central" + else: + scraper_type = "release" + if channel == "stable": + version = "latest" + else: + assert channel == "esr" + version = "latest-esr" + kwargs["version"] = version + + logger.info("downloading Firefox %s...", channel) + scraper = mozdownload.FactoryScraper( + scraper_type, + platform=mozdownload_platform, + destination=str(install_dir), + **kwargs, + ) + archive = scraper.download() + + install_path = mozinstall.install(archive, str(install_dir)) + binary = Path(mozinstall.get_binary(install_path, "firefox")) + + logger.info("installed Firefox at %s", binary) + return binary + + +class FirefoxBrowsers: + def __init__(self) -> None: + self._nightly: Path | None = None + self._esr: Path | None = None + self._stable: Path | None = None + + @property + def nightly(self) -> Path: + if self._nightly is None: + self._nightly = install_firefox(channel="nightly") + return self._nightly + + @property + def stable(self) -> Path: + if self._stable is None: + self._stable = install_firefox(channel="stable") + return self._stable + + @property + def esr(self) -> Path: + if self._esr is None: + self._esr = install_firefox(channel="esr") + return self._esr diff --git a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/firefox_install.py b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/firefox_install.py deleted file mode 100644 index 93f63475da..0000000000 --- a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/firefox_install.py +++ /dev/null @@ -1,45 +0,0 @@ -"""Download and install a prebuilt Firefox Nightly for the agent to drive.""" - -from __future__ import annotations - -import logging -import platform -import shutil -from pathlib import Path - -import mozdownload -import mozinstall - -# Directory to install into, and the mozdownload branch to pull the daily build from -INSTALL_DIR = Path.home() / "firefox" -BRANCH = "mozilla-central" - -logger = logging.getLogger("autowebcompat-repro") - - -def install_firefox_nightly() -> Path: - # mozdownload guesses the platform from OS + bit-width only, ignoring CPU arch — - # so on 64-bit Linux it always picks the x86-64 build, even on ARM. Override to - # the ARM build on ARM hosts; pass None elsewhere to let it auto-detect. - mozdownload_platform = ( - "linux-arm64" if platform.machine() in ("aarch64", "arm64") else None - ) - - if INSTALL_DIR.exists(): - shutil.rmtree(INSTALL_DIR) - INSTALL_DIR.mkdir(parents=True) - - logger.info("downloading Firefox Nightly...") - scraper = mozdownload.FactoryScraper( - "daily", - branch=BRANCH, - platform=mozdownload_platform, - destination=str(INSTALL_DIR), - ) - archive = scraper.download() - - install_folder = mozinstall.install(archive, str(INSTALL_DIR)) - binary = Path(mozinstall.get_binary(install_folder, "firefox")) - - logger.info("installed Firefox at %s", binary) - return binary diff --git a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/result.py b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/result.py index fdb3733c67..4450428077 100644 --- a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/result.py +++ b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/result.py @@ -23,51 +23,71 @@ def __init__(self, result_cls: type[ResultT]) -> None: self.result: ResultT | None = None -class ReproductionResult(BaseModel): - """Canonical result the agent produces for a web-compat investigation.""" +class TestPlanResult(BaseModel): + is_webcompat: bool = Field( + description=("true if the input describes a webcompat issue, otherwise false."), + ) + + affects_platforms: list[ + Literal["ios"] | Literal["android"] | Literal["desktop"] + ] = Field(description="List of platforms which seem to be affected by the issue") + + affects_os: ( + None + | Literal["all"] + | list[Literal["windows"] | Literal["linux"] | Literal["macos"]] + ) = Field( + description="""List of desktop issues known to be affected. + - `null` if the issue does not affect desktop. + - "all" if there is no strong evidence that the issue is platform specific" + - Otherwise a list of platform names which are likely affected + """ + ) + affects_channels: list[Literal["nightly"] | Literal["stable"] | Literal["esr"]] = ( + Field( + description="""List of channels affected + - "esr" if the issue is reported as specific to ESR builds. + - "stable" if the issue is reported as reproducing on stable builds, or there is no evidence for which channels are affected + - "nightly" if the issue is reported as reproducing on nightly builds, or there is no evidence for which channels are affected + """ + ) + ) + + +class ReproductionResult(BaseModel): reproduced: bool = Field( description=( "true if the reported issue reproduced in Firefox, otherwise false." ), ) - summary: str = Field( - description="""A concise account of whether the issue represents a real - webcompat issue i.e. it can be reproduced in Firefox.""" - ) failure_reason: ( - Literal["not_reproduced"] + Literal["not_reproducable"] | Literal["non_compat"] + | Literal["unsupported_platform"] | Literal["blocked"] + | Literal["blocked_captcha"] + | Literal["blocked_geo"] | Literal["login"] | Literal["down"] | Literal["other"] | None ) = Field( - description="""When an issue could not be reproduced, one of + description="""If an issue was reproduced then `null`. When an issue could not be reproduced, one of following categories describing the reason for the failure: - * not_reproduced - When it was possible to run all the steps to reproduce, but no issue was found + * not_reproducable - When it was possible to run all the steps to reproduce, but no issue was found * non_compat - When the report doesn't refer to site breakage for example for issues with the Firefox UI or product features such as reader mode - * blocked - When access to the site was blocked (e.g. due to geoblocking or because the page requires solving a captcha) + * unsupported_platform - When the report is specific to a platform that isn't available e.g. iOS + * blocked_captcha - When access to the site was blocked because the page requires solving a captcha + * blocked_geo - When access to the site was blocked based on location ("geoblocking") + * blocked - When access to the site was blocked for some reason that couldn't be identified as a captcha or geoblocking * login - When reproducing the issue requires completing a login flow - * down - Site down or unavailable + * down - When the site down or unavailable in a way that is unrelated to the issue report * other - When the issue could not be reproduced for some other reason (please give details in the summary text) -""", - ) - steps: str = Field( - description=( - "The ordered steps you took, as a single numbered list (1., 2., 3., " - "... one step per line), written so another agent could reproduce " - "them with no extra context. Each step must be self-contained: " - "whenever you introduce an input or artifact the report did not " - "provide (a file, image, account, or any other test data), state its " - "exact origin — the URL you fetched it from, the command you ran, or " - 'how you generated it — not just that you "used" or "saved" it. A ' - "reader must be able to obtain the same inputs. Omit the reproduction " - "screenshot step." - ), +""" ) + screenshot_path: Path | None = Field( description=( """The file path you saved a screenshot to via the `screenshot_page` @@ -91,6 +111,29 @@ def validate_screenshot_path(cls, path: Path | None) -> Path | None: return path +class BugReproductionResult(ReproductionResult): + """Canonical result the agent produces for a web-compat investigation.""" + + summary: str = Field( + description="""A concise account of whether the issue represents a real + webcompat issue i.e. it can be reproduced in Firefox.""" + ) + + steps: str = Field( + description=( + "The ordered steps you took, as a single numbered list (1., 2., 3., " + "... one step per line), written so another agent could reproduce " + "them with no extra context. Each step must be self-contained: " + "whenever you introduce an input or artifact the report did not " + "provide (a file, image, account, or any other test data), state its " + "exact origin — the URL you fetched it from, the command you ran, or " + 'how you generated it — not just that you "used" or "saved" it. A ' + "reader must be able to obtain the same inputs. Omit the reproduction " + "screenshot step." + ), + ) + + class ChromeMaskResult(BaseModel): chrome_mask_fixed: bool | None = Field( description=( From b117f5ae97f335a33a69e3fdf523deed85c4fb4d Mon Sep 17 00:00:00 2001 From: James Graham Date: Fri, 10 Jul 2026 11:24:05 +0100 Subject: [PATCH 2/2] Fix type annotations in broker --- .../hackbot_agents/autowebcompat_repro/broker.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/broker.py b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/broker.py index c071cb3953..2a17f77d15 100644 --- a/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/broker.py +++ b/agents/autowebcompat-repro/hackbot_agents/autowebcompat_repro/broker.py @@ -8,13 +8,19 @@ import logging from contextlib import asynccontextmanager +from typing import AsyncIterator import bugsy import uvicorn from agent_tools import bugzilla from agent_tools.bugzilla import BugzillaContext from agent_tools.claude_sdk import build_sdk_server -from mcp.server.streamable_http_manager import StreamableHTTPSessionManager +from mcp.server.streamable_http_manager import ( + Receive, + Scope, + Send, + StreamableHTTPSessionManager, +) from pydantic_settings import BaseSettings, SettingsConfigDict from starlette.applications import Starlette from starlette.routing import Mount @@ -42,7 +48,7 @@ def build_app(inputs: BrokerInputs) -> Starlette: manager = StreamableHTTPSessionManager(app=mcp_server, stateless=True) @asynccontextmanager - async def lifespan(app): + async def lifespan(app: Starlette) -> AsyncIterator[None]: async with manager.run(): log.info( "bugzilla broker ready on %s:%d (read-only)", @@ -51,7 +57,7 @@ async def lifespan(app): ) yield - async def mcp_handler(scope, receive, send): + async def mcp_handler(scope: Scope, receive: Receive, send: Send) -> None: await manager.handle_request(scope, receive, send) return Starlette(routes=[Mount("/mcp", app=mcp_handler)], lifespan=lifespan)