From 9a000d7b1138b6be1f229a0a37a99f29f03b8830 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 22:46:18 -0500 Subject: [PATCH 1/4] perf: fix build errors concurrently with bounded fan-out fix_build_errors repaired each failing file in a strictly serial loop, performing blocking Path.read_text/write_text on the event loop thread around each LLM round-trip. The files are mutually independent, so wall clock scaled linearly with the number of broken files. - offload read_text/write_text via asyncio.to_thread - fan out per-file fixes with asyncio.gather, bounded by a semaphore (default 4, overridable via the new max_concurrency parameter) - clamp non-positive max_concurrency to 1 - sort fixed_files for deterministic results (error_files is a set) Failure semantics are unchanged: a missing, unreadable, or provider-failed file is skipped without aborting its siblings. DeploymentManager.verify_and_fix_project calls this from inside a retry loop, so the serial cost was multiplied by max_retries per deployment. Closes #1335 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/ai_code_generator.py | 68 ++++++-- tests/unit/test_ai_code_generator.py | 160 ++++++++++++++++++ 2 files changed, 218 insertions(+), 10 deletions(-) diff --git a/src/youtube_extension/backend/ai_code_generator.py b/src/youtube_extension/backend/ai_code_generator.py index 840827094..4b36d3647 100644 --- a/src/youtube_extension/backend/ai_code_generator.py +++ b/src/youtube_extension/backend/ai_code_generator.py @@ -8,6 +8,7 @@ """ import ast +import asyncio import json import logging import os @@ -18,6 +19,11 @@ logger = logging.getLogger(__name__) +# Upper bound on concurrent AI auto-fix calls. Each fix is an independent +# multi-second LLM round-trip, so some fan-out is a large win, but an unbounded +# one would hit provider rate limits on projects with many failing files. +_MAX_CONCURRENT_FIXES = 4 + # Add project root for knowledge_base import # Add scripts directory for knowledge_base import # ai_code_generator.py is in src/youtube_extension/backend/ @@ -1260,15 +1266,21 @@ async def fix_build_errors( self, project_path: Path, errors: list[str], - suggested_fixes: list[str] + suggested_fixes: list[str], + max_concurrency: Optional[int] = None ) -> dict[str, Any]: """ Use AI to fix build errors in generated code. + Files are fixed concurrently, bounded by *max_concurrency*, because each + one is independent and dominated by a network round-trip to the LLM. + Args: project_path: Path to the project errors: List of build error messages suggested_fixes: List of suggested resolutions from skill database + max_concurrency: Max files fixed in parallel. Defaults to + _MAX_CONCURRENT_FIXES; values below 1 are clamped to 1. Returns: Dict with fixed files and status @@ -1293,14 +1305,37 @@ async def fix_build_errors( # Try common problem files error_files = {"src/app/page.tsx", "src/components/Button.tsx"} - fixed_files = [] - for rel_path in error_files: + # Each error file is independent: its own read, its own AI call and its + # own write. self.router.generate() is a multi-second network round-trip, + # so running them one after another made wall-clock cost scale linearly + # with the number of failing files. Bound the fan-out so we do not trip + # provider rate limits. + limit = ( + _MAX_CONCURRENT_FIXES + if max_concurrency is None + else max(1, max_concurrency) + ) + semaphore = asyncio.Semaphore(limit) + + async def _fix_one(rel_path: str) -> Optional[str]: + """Fix one file; return its path on success, otherwise None.""" file_path = project_path / rel_path - if not file_path.exists(): - continue - # Read current file content - current_content = file_path.read_text() + def _read_source() -> Optional[str]: + # Path.exists()/read_text() are blocking syscalls - keep them + # off the event loop. + if not file_path.exists(): + return None + return file_path.read_text() + + try: + current_content = await asyncio.to_thread(_read_source) + except OSError as e: + logger.warning(f"⚠️ Failed to read {rel_path}: {e}") + return None + + if current_content is None: + return None # Build fix prompt fix_prompt = f"""You are a TypeScript/Next.js expert. Fix the following code that has build errors. @@ -1328,7 +1363,7 @@ async def fix_build_errors( if not response_text: logger.warning(f"⚠️ LLM router returned no text for {rel_path}, skipping") - continue + return None fixed_code = response_text.strip() @@ -1346,12 +1381,25 @@ async def fix_build_errors( break # Write fixed content - file_path.write_text(fixed_code) - fixed_files.append(rel_path) + await asyncio.to_thread(file_path.write_text, fixed_code) logger.info(f"✅ Fixed: {rel_path}") + return rel_path except Exception as e: logger.warning(f"⚠️ Failed to fix {rel_path}: {e}") + return None + + async def _fix_guarded(rel_path: str) -> Optional[str]: + async with semaphore: + return await _fix_one(rel_path) + + # error_files is a set, so the previous sequential loop reported results + # in arbitrary order; sort for a stable, assertable ordering. + ordered_paths = sorted(error_files) + results = await asyncio.gather( + *(_fix_guarded(rel_path) for rel_path in ordered_paths) + ) + fixed_files = [rel_path for rel_path in results if rel_path] return { "success": len(fixed_files) > 0, diff --git a/tests/unit/test_ai_code_generator.py b/tests/unit/test_ai_code_generator.py index f2884b06c..b0d9e4a99 100644 --- a/tests/unit/test_ai_code_generator.py +++ b/tests/unit/test_ai_code_generator.py @@ -2,8 +2,10 @@ from __future__ import annotations +import asyncio import json import sys +import threading from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -1073,6 +1075,164 @@ async def test_none_response_skips_file(self, tmp_path): assert result["success"] is False +class TestFixBuildErrorsConcurrency: + """fix_build_errors fans files out concurrently and keeps I/O off the loop. + + Each error file is independent (own read, own LLM round-trip, own write), so + the previous sequential loop made wall-clock cost scale linearly with the + number of failing files while the event loop sat blocked on file I/O. + """ + + @staticmethod + def _make_files(tmp_path, count): + rel_paths = [] + for i in range(count): + rel = f"src/app/page{i}.tsx" + file_path = tmp_path / rel + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text(f"const x = {i};") + rel_paths.append(rel) + return rel_paths + + @staticmethod + def _errors_for(rel_paths): + return [f"error in {rel}:1:5 - Type error" for rel in rel_paths] + + @staticmethod + def _tracking_generate(record): + """Return an async generate() that records peak concurrent calls.""" + state = {"inflight": 0} + + async def _generate(prompt, **kwargs): + state["inflight"] += 1 + record["max_inflight"] = max( + record.get("max_inflight", 0), state["inflight"] + ) + try: + await asyncio.sleep(0.05) + return "const fixed = true;" + finally: + state["inflight"] -= 1 + + return _generate + + async def test_files_are_fixed_concurrently(self, tmp_path): + gen = _make_gen_with_mock_client(tmp_path) + rel_paths = self._make_files(tmp_path, 4) + record = {} + gen.router.generate = AsyncMock(side_effect=self._tracking_generate(record)) + + result = await gen.fix_build_errors(tmp_path, self._errors_for(rel_paths), []) + + assert result["success"] is True + assert len(result["fixed_files"]) == 4 + # A sequential loop can never have more than one call in flight. + assert record["max_inflight"] > 1 + + async def test_default_concurrency_is_bounded(self, tmp_path): + gen = _make_gen_with_mock_client(tmp_path) + rel_paths = self._make_files(tmp_path, 8) + record = {} + gen.router.generate = AsyncMock(side_effect=self._tracking_generate(record)) + + await gen.fix_build_errors(tmp_path, self._errors_for(rel_paths), []) + + # Unbounded fan-out would reach 8 and risk provider rate limits. + assert 1 < record["max_inflight"] <= _mod._MAX_CONCURRENT_FIXES + + async def test_max_concurrency_override_is_respected(self, tmp_path): + gen = _make_gen_with_mock_client(tmp_path) + rel_paths = self._make_files(tmp_path, 8) + record = {} + gen.router.generate = AsyncMock(side_effect=self._tracking_generate(record)) + + result = await gen.fix_build_errors( + tmp_path, self._errors_for(rel_paths), [], max_concurrency=2 + ) + + assert len(result["fixed_files"]) == 8 + assert record["max_inflight"] == 2 + + @pytest.mark.parametrize("bad_limit", [0, -5]) + async def test_non_positive_max_concurrency_clamps_to_one(self, tmp_path, bad_limit): + gen = _make_gen_with_mock_client(tmp_path) + rel_paths = self._make_files(tmp_path, 3) + record = {} + gen.router.generate = AsyncMock(side_effect=self._tracking_generate(record)) + + result = await gen.fix_build_errors( + tmp_path, self._errors_for(rel_paths), [], max_concurrency=bad_limit + ) + + assert len(result["fixed_files"]) == 3 + assert record["max_inflight"] == 1 + + async def test_fixed_files_order_is_deterministic(self, tmp_path): + """error_files is a set, so results must be sorted to be assertable.""" + gen = _make_gen_with_mock_client(tmp_path) + rel_paths = self._make_files(tmp_path, 5) + gen.router.generate = AsyncMock(return_value="const fixed = true;") + + result = await gen.fix_build_errors(tmp_path, self._errors_for(rel_paths), []) + + assert result["fixed_files"] == sorted(rel_paths) + + async def test_file_io_runs_off_the_event_loop_thread(self, tmp_path, monkeypatch): + gen = _make_gen_with_mock_client(tmp_path) + rel_paths = self._make_files(tmp_path, 2) + gen.router.generate = AsyncMock(return_value="const fixed = true;") + + loop_thread = threading.get_ident() + read_threads: list[int] = [] + write_threads: list[int] = [] + real_read, real_write = Path.read_text, Path.write_text + + def _tracked_read(self, *args, **kwargs): + read_threads.append(threading.get_ident()) + return real_read(self, *args, **kwargs) + + def _tracked_write(self, *args, **kwargs): + write_threads.append(threading.get_ident()) + return real_write(self, *args, **kwargs) + + monkeypatch.setattr(Path, "read_text", _tracked_read) + monkeypatch.setattr(Path, "write_text", _tracked_write) + + await gen.fix_build_errors(tmp_path, self._errors_for(rel_paths), []) + + assert read_threads and write_threads + assert loop_thread not in read_threads + assert loop_thread not in write_threads + + async def test_one_file_failure_does_not_abort_siblings(self, tmp_path): + gen = _make_gen_with_mock_client(tmp_path) + rel_paths = self._make_files(tmp_path, 3) + + async def _generate(prompt, **kwargs): + # The prompt embeds the whole error list, so key off the per-file + # "CURRENT CODE ()" marker to target exactly one file. + if "CURRENT CODE (src/app/page1.tsx)" in prompt: + raise RuntimeError("provider blew up") + return "const fixed = true;" + + gen.router.generate = AsyncMock(side_effect=_generate) + + result = await gen.fix_build_errors(tmp_path, self._errors_for(rel_paths), []) + + assert result["fixed_files"] == ["src/app/page0.tsx", "src/app/page2.tsx"] + assert result["success"] is True + + async def test_missing_file_does_not_abort_siblings(self, tmp_path): + gen = _make_gen_with_mock_client(tmp_path) + rel_paths = self._make_files(tmp_path, 2) + gen.router.generate = AsyncMock(return_value="const fixed = true;") + errors = self._errors_for([*rel_paths, "src/app/gone.tsx"]) + + result = await gen.fix_build_errors(tmp_path, errors, []) + + assert result["fixed_files"] == sorted(rel_paths) + + # =========================================================================== # AICodeGenerator._generate_turborepo_monorepo # =========================================================================== From b57b22b2c4227fc49d535858a2f45216f99ac3ff Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 22:56:29 -0500 Subject: [PATCH 2/4] fix: share the AI fix budget process-wide and isolate decode errors Addresses both review findings on #1336. 1. UnicodeDecodeError escaped the read guard. Path.read_text() raises UnicodeDecodeError on a non-UTF-8 source. That is a ValueError subclass, not an OSError, so it slipped past `except OSError` and out of _fix_one. With gather(return_exceptions=False) that does not merely lose one file -- it aborts the entire fan-out and discards every sibling's already-completed fix. The guard now catches (OSError, UnicodeError). 2. The concurrency bound was per-invocation, not process-wide. get_deployment_manager() constructs a fresh DeploymentManager, and therefore a fresh AICodeGenerator, for every pipeline run. A semaphore owned by a single fix_build_errors call let M concurrent deployments issue M * limit LLM calls, defeating the rate-limit protection the bound exists to provide. _shared_fix_semaphore() now returns one semaphore per (running loop, limit). Keying on the loop is required because asyncio primitives bind to the first loop that awaits them, so a plain module-level singleton would raise "bound to a different event loop" as soon as a second loop used it. A WeakKeyDictionary keeps finished loops from leaking, and a TypeError fallback covers loop implementations that are not weak-referenceable. Both fixes are prove-failed: each new test fails against the previous commit and passes here. 252 passed (was 250). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/ai_code_generator.py | 54 ++++++++++++++-- tests/unit/test_ai_code_generator.py | 61 +++++++++++++++++++ 2 files changed, 110 insertions(+), 5 deletions(-) diff --git a/src/youtube_extension/backend/ai_code_generator.py b/src/youtube_extension/backend/ai_code_generator.py index 4b36d3647..89c45097b 100644 --- a/src/youtube_extension/backend/ai_code_generator.py +++ b/src/youtube_extension/backend/ai_code_generator.py @@ -13,6 +13,7 @@ import logging import os import sys +import weakref from datetime import datetime from pathlib import Path from typing import Any, Optional @@ -24,6 +25,41 @@ # one would hit provider rate limits on projects with many failing files. _MAX_CONCURRENT_FIXES = 4 +# The fix budget is *process-wide*, not per-invocation. get_deployment_manager() +# builds a fresh DeploymentManager -- and therefore a fresh AICodeGenerator -- +# for every pipeline run, so a semaphore owned by one call would let N concurrent +# deployments issue N * limit LLM calls and defeat the rate-limit protection this +# bound exists to provide. Sharing one semaphore per (event loop, limit) caps +# total in-flight fixes at `limit` no matter how many generators exist. +# +# Keyed by the running loop because asyncio primitives bind to the first loop +# that awaits them; a module-level singleton would raise once a second loop +# (e.g. the next test case) tried to use it. WeakKeyDictionary so finished +# loops do not leak. +_FIX_SEMAPHORES: "weakref.WeakKeyDictionary[Any, dict[int, asyncio.Semaphore]]" = ( + weakref.WeakKeyDictionary() +) + + +def _shared_fix_semaphore(limit: int) -> asyncio.Semaphore: + """Return the process-wide fix semaphore for `limit` on the running loop.""" + loop = asyncio.get_running_loop() + try: + by_limit = _FIX_SEMAPHORES.get(loop) + if by_limit is None: + by_limit = {} + _FIX_SEMAPHORES[loop] = by_limit + except TypeError: + # Some loop implementations are not weak-referenceable. Fall back to an + # unshared semaphore: the per-invocation bound still holds. + return asyncio.Semaphore(limit) + + semaphore = by_limit.get(limit) + if semaphore is None: + semaphore = asyncio.Semaphore(limit) + by_limit[limit] = semaphore + return semaphore + # Add project root for knowledge_base import # Add scripts directory for knowledge_base import # ai_code_generator.py is in src/youtube_extension/backend/ @@ -1273,14 +1309,17 @@ async def fix_build_errors( Use AI to fix build errors in generated code. Files are fixed concurrently, bounded by *max_concurrency*, because each - one is independent and dominated by a network round-trip to the LLM. + one is independent and dominated by a network round-trip to the LLM. The + bound is shared process-wide, so concurrent deployments cannot multiply + it (see _shared_fix_semaphore). Args: project_path: Path to the project errors: List of build error messages suggested_fixes: List of suggested resolutions from skill database - max_concurrency: Max files fixed in parallel. Defaults to - _MAX_CONCURRENT_FIXES; values below 1 are clamped to 1. + max_concurrency: Max files fixed in parallel, across *all* in-flight + calls in this process. Defaults to _MAX_CONCURRENT_FIXES; values + below 1 are clamped to 1. Returns: Dict with fixed files and status @@ -1315,7 +1354,7 @@ async def fix_build_errors( if max_concurrency is None else max(1, max_concurrency) ) - semaphore = asyncio.Semaphore(limit) + semaphore = _shared_fix_semaphore(limit) async def _fix_one(rel_path: str) -> Optional[str]: """Fix one file; return its path on success, otherwise None.""" @@ -1330,7 +1369,12 @@ def _read_source() -> Optional[str]: try: current_content = await asyncio.to_thread(_read_source) - except OSError as e: + except (OSError, UnicodeError) as e: + # UnicodeDecodeError (a ValueError, *not* an OSError) is raised by + # read_text() on a non-UTF-8 source. Letting it escape _fix_one + # would propagate through gather(return_exceptions=False) and throw + # away every sibling file's successful fix, so it is handled here + # as just another per-file read failure. logger.warning(f"⚠️ Failed to read {rel_path}: {e}") return None diff --git a/tests/unit/test_ai_code_generator.py b/tests/unit/test_ai_code_generator.py index b0d9e4a99..e31ef509a 100644 --- a/tests/unit/test_ai_code_generator.py +++ b/tests/unit/test_ai_code_generator.py @@ -1232,6 +1232,67 @@ async def test_missing_file_does_not_abort_siblings(self, tmp_path): assert result["fixed_files"] == sorted(rel_paths) + async def test_undecodable_source_does_not_abort_siblings(self, tmp_path): + """A non-UTF-8 source must be skipped, not blow up the whole fan-out. + + Path.read_text() raises UnicodeDecodeError -- a ValueError subclass, *not* + an OSError -- on undecodable bytes. Because the files are gathered with + the default return_exceptions=False, letting that escape _fix_one would + propagate out of gather and discard every sibling's successful fix. + """ + gen = _make_gen_with_mock_client(tmp_path) + rel_paths = self._make_files(tmp_path, 3) + # Invalid UTF-8: a lone continuation byte cannot start a sequence. + (tmp_path / "src/app/page1.tsx").write_bytes(b"const x = '\xff\xfe';") + gen.router.generate = AsyncMock(return_value="const fixed = true;") + + result = await gen.fix_build_errors(tmp_path, self._errors_for(rel_paths), []) + + assert result["fixed_files"] == ["src/app/page0.tsx", "src/app/page2.tsx"] + assert result["success"] is True + + async def test_concurrent_invocations_share_the_fix_budget(self, tmp_path): + """The concurrency bound is process-wide, not per-invocation. + + get_deployment_manager() builds a fresh DeploymentManager -- and so a + fresh AICodeGenerator -- per pipeline run. A semaphore owned by a single + call would therefore let N concurrent deployments issue N * limit LLM + calls, defeating the rate-limit protection the bound exists to provide. + """ + record = {"max_inflight": 0} + shared = {"inflight": 0} + + async def _generate(prompt, **kwargs): + shared["inflight"] += 1 + record["max_inflight"] = max(record["max_inflight"], shared["inflight"]) + try: + await asyncio.sleep(0.05) + return "const fixed = true;" + finally: + shared["inflight"] -= 1 + + projects = [] + for n in range(3): + root = tmp_path / f"proj{n}" + root.mkdir() + rel_paths = self._make_files(root, 4) + gen = _make_gen_with_mock_client(root) + gen.router.generate = AsyncMock(side_effect=_generate) + projects.append((gen, root, rel_paths)) + + results = await asyncio.gather( + *( + gen.fix_build_errors(root, self._errors_for(rel_paths), []) + for gen, root, rel_paths in projects + ) + ) + + # All 12 files across all 3 generators still get fixed... + assert [len(r["fixed_files"]) for r in results] == [4, 4, 4] + # ...but the three independent generators never exceed one shared budget. + assert record["max_inflight"] > 1 + assert record["max_inflight"] <= 4 + # =========================================================================== # AICodeGenerator._generate_turborepo_monorepo From 370bcec537dc961029f57f28d49c67b265aea2ba Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 23:07:37 -0500 Subject: [PATCH 3/4] fix: drain the in-flight file write before propagating cancellation asyncio.to_thread cannot interrupt a worker thread, so cancelling a bare `await asyncio.to_thread(file_path.write_text, ...)` returns control to the caller while the thread is still truncating and rewriting the file. gather() cancels every sibling as soon as one task raises, so this is reachable in normal operation: the caller sees a failed repair and may start cleanup on a file that is actively being written. Shield the write so cancellation cannot detach it, then drain it before re-raising. Exceptions from the drain are suppressed so a failing write cannot swallow the cancellation. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/ai_code_generator.py | 21 +++++++- tests/unit/test_ai_code_generator.py | 49 +++++++++++++++++++ 2 files changed, 68 insertions(+), 2 deletions(-) diff --git a/src/youtube_extension/backend/ai_code_generator.py b/src/youtube_extension/backend/ai_code_generator.py index 89c45097b..426f81803 100644 --- a/src/youtube_extension/backend/ai_code_generator.py +++ b/src/youtube_extension/backend/ai_code_generator.py @@ -9,6 +9,7 @@ import ast import asyncio +import contextlib import json import logging import os @@ -1424,8 +1425,24 @@ def _read_source() -> Optional[str]: fixed_code = block break - # Write fixed content - await asyncio.to_thread(file_path.write_text, fixed_code) + # Write fixed content. asyncio.to_thread cannot interrupt the + # worker thread once write_text has started, so a plain + # `await asyncio.to_thread(...)` that gets cancelled -- which + # gather() does to every sibling as soon as one task raises -- + # would return control to the caller while the thread is still + # writing. The caller would then see a "failed" repair and could + # start cleanup on a file being truncated and rewritten + # underneath it. Shield the write so cancellation doesn't detach + # it, then drain it before propagating. + write = asyncio.create_task( + asyncio.to_thread(file_path.write_text, fixed_code) + ) + try: + await asyncio.shield(write) + except asyncio.CancelledError: + with contextlib.suppress(Exception): + await write + raise logger.info(f"✅ Fixed: {rel_path}") return rel_path diff --git a/tests/unit/test_ai_code_generator.py b/tests/unit/test_ai_code_generator.py index e31ef509a..819164f9c 100644 --- a/tests/unit/test_ai_code_generator.py +++ b/tests/unit/test_ai_code_generator.py @@ -1293,6 +1293,55 @@ async def _generate(prompt, **kwargs): assert record["max_inflight"] > 1 assert record["max_inflight"] <= 4 + async def test_cancellation_drains_the_in_flight_write( + self, tmp_path, monkeypatch + ): + """A cancelled fix must not abandon a half-written file. + + asyncio.to_thread hands work to a worker thread and has no way to + interrupt it. Cancelling a bare `await asyncio.to_thread(write_text, ...)` + therefore returns control to the caller immediately while the thread is + still truncating and rewriting the file -- and gather() cancels every + sibling the moment one task raises, so this is reachable in normal + operation, not just on Ctrl-C. The caller would see a failed repair and + could begin cleanup on a file that is actively being written. + """ + gen = _make_gen_with_mock_client(tmp_path) + rel_paths = self._make_files(tmp_path, 1) + target = tmp_path / rel_paths[0] + original = target.read_text() + gen.router.generate = AsyncMock(return_value="const fixed = true;") + + write_started = threading.Event() + real_write = Path.write_text + + def _slow_write(self, *args, **kwargs): + if self == target: + write_started.set() + # Hold the worker thread open long enough for the test to + # cancel while this write is genuinely mid-flight. + threading.Event().wait(0.3) + return real_write(self, *args, **kwargs) + + monkeypatch.setattr(Path, "write_text", _slow_write) + + task = asyncio.create_task( + gen.fix_build_errors(tmp_path, self._errors_for(rel_paths), []) + ) + # Wait for the worker thread to actually enter write_text before + # cancelling, so the race window is open rather than assumed. + await asyncio.to_thread(write_started.wait, 5) + assert write_started.is_set() + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + + # Cancellation propagated only after the write was drained, so the file + # is whole. Without the drain this still holds the pre-fix contents. + assert target.read_text() == "const fixed = true;" + assert target.read_text() != original + # =========================================================================== # AICodeGenerator._generate_turborepo_monorepo From af277dd530761c9e8425ec8181de463dbfdf39bd Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 23:10:39 -0500 Subject: [PATCH 4/4] test: make the cancellation drain test deterministic The worker thread now blocks on an explicit Event that the test releases only after cancelling, instead of sleeping for a fixed interval. A fixed sleep can elapse before the cancellation lands on a loaded runner, which would let the test pass without ever opening the race window it exists to prove. Verified 3/3 pass with the drain and 3/3 fail without it. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/unit/test_ai_code_generator.py | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/tests/unit/test_ai_code_generator.py b/tests/unit/test_ai_code_generator.py index 819164f9c..b28a50a8c 100644 --- a/tests/unit/test_ai_code_generator.py +++ b/tests/unit/test_ai_code_generator.py @@ -1313,14 +1313,16 @@ async def test_cancellation_drains_the_in_flight_write( gen.router.generate = AsyncMock(return_value="const fixed = true;") write_started = threading.Event() + may_finish = threading.Event() real_write = Path.write_text def _slow_write(self, *args, **kwargs): if self == target: write_started.set() - # Hold the worker thread open long enough for the test to - # cancel while this write is genuinely mid-flight. - threading.Event().wait(0.3) + # Block until the test explicitly releases us. A fixed sleep + # would let this write complete early on a slow runner and the + # test would pass without ever opening the race window. + may_finish.wait(5) return real_write(self, *args, **kwargs) monkeypatch.setattr(Path, "write_text", _slow_write) @@ -1334,8 +1336,18 @@ def _slow_write(self, *args, **kwargs): assert write_started.is_set() task.cancel() - with pytest.raises(asyncio.CancelledError): - await task + # Release the worker only after the cancellation has been delivered. + # With the drain the coroutine is parked on the write and observes it + # complete; without it the coroutine has already unwound and the file + # is still unwritten when the assertion below runs. + releaser = threading.Timer(0.2, may_finish.set) + releaser.start() + try: + with pytest.raises(asyncio.CancelledError): + await task + finally: + releaser.cancel() + may_finish.set() # Cancellation propagated only after the write was drained, so the file # is whole. Without the drain this still holds the pre-fix contents.