try_fixes.py
The same, with the async client and every fork running at once.
import asyncioimport osimport sys from upstream import AsyncUpstream TEST = "cd app && git apply fix.patch && uv run pytest -q" async def attempt(sb, patch): await sb.write_file("/workspace/app/fix.patch", patch) r = await sb.exec(["sh", "-c", TEST]) return r.exit_code == 0 async def main(repo, patches): async with AsyncUpstream(api_key=os.environ["UPSTREAM_API_KEY"]) as client: base = await client.sandbox(template="python") await base.exec(["sh", "-c", f"git clone {repo} app && cd app && uv sync"]) forks = [await base.fork() for _ in patches] passed = await asyncio.gather(*(attempt(sb, p) for sb, p in zip(forks, patches))) for sb, ok in zip(forks, passed): print(sb.sandbox_id, "pass" if ok else "fail") await sb.destroy() await base.destroy() if __name__ == "__main__": repo, *patch_files = sys.argv[1:] asyncio.run(main(repo, [open(p).read() for p in patch_files]))