Skip to content

Instantly share code, notes, and snippets.

@eric239

eric239/pr-4639-review.html

Created April 2, 2026 16:11
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save eric239/6df3d6e43f07eeb674eedfd0db3c60ec to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save eric239/6df3d6e43f07eeb674eedfd0db3c60ec to your computer and use it in GitHub Desktop.
Download ZIP
PR #4639 agent review report
Raw
pr-4639-review.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>PR #4639 Review - TECH-6312: Asyncio infrastructure for prefork Celery workers</title>
<style>
:root {
--bg: #0d1117;
--surface: #161b22;
--border: #30363d;
--text: #e6edf3;
--text-muted: #8b949e;
--accent: #58a6ff;
--green: #3fb950;
--red: #f85149;
--green-bg: #12261e;
--red-bg: #2d1214;
--yellow: #d29922;
--purple: #bc8cff;
--orange: #f0883e;
--syn-keyword: #ff7b72;
--syn-string: #a5d6ff;
--syn-decorator: #d2a8ff;
--syn-func: #d2a8ff;
--syn-class: #ffa657;
--syn-builtin: #ffa657;
--syn-comment: #8b949e;
--syn-self: #ff7b72;
--syn-number: #79c0ff;
--syn-type: #ffa657;
}
* { margin: 0; padding: 0; box-sizing: border-box; }
body { background: var(--bg); color: var(--text); font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Helvetica, Arial, sans-serif; line-height: 1.6; padding: 2rem; max-width: 1200px; margin: 0 auto; }
a { color: var(--accent); text-decoration: none; }
a:hover { text-decoration: underline; }
.header { border-bottom: 1px solid var(--border); padding-bottom: 1.5rem; margin-bottom: 2rem; }
.header h1 { font-size: 1.5rem; font-weight: 600; margin-bottom: 0.5rem; }
.header .meta { color: var(--text-muted); font-size: 0.85rem; display: flex; gap: 1.5rem; flex-wrap: wrap; }
.header .meta span { display: inline-flex; align-items: center; gap: 0.3rem; }
.tldr { border-left: 3px solid var(--accent); background: rgba(88,166,255,0.08); padding: 1rem 1.25rem; margin-bottom: 2rem; border-radius: 0 6px 6px 0; }
.tldr p { font-size: 0.92rem; }
.diagram-container { margin-bottom: 2rem; background: var(--surface); border: 1px solid var(--border); border-radius: 8px; padding: 1.5rem; text-align: center; }
.block-card { background: var(--surface); border: 1px solid var(--border); border-radius: 8px; margin-bottom: 1.5rem; overflow: hidden; }
.block-header { padding: 1rem 1.25rem; border-bottom: 1px solid var(--border); display: flex; align-items: center; gap: 0.75rem; }
.block-header .num { background: var(--accent); color: #fff; width: 28px; height: 28px; border-radius: 50%; display: flex; align-items: center; justify-content: center; font-size: 0.8rem; font-weight: 700; flex-shrink: 0; }
.block-header h2 { font-size: 1.1rem; font-weight: 600; }
.block-body { display: flex; }
.block-files { width: 220px; flex-shrink: 0; background: rgba(22,27,34,0.8); padding: 1rem 1.25rem; border-right: 1px solid var(--border); }
.block-files .file { display: flex; align-items: center; gap: 0.5rem; font-size: 0.78rem; font-family: 'SF Mono', SFMono-Regular, Consolas, monospace; padding: 0.2rem 0; word-break: break-all; }
.block-files .dot { width: 8px; height: 8px; border-radius: 50%; flex-shrink: 0; }
.dot-added { background: var(--green); }
.dot-modified { background: var(--yellow); }
.dot-removed { background: var(--red); }
.block-explanation { padding: 1rem 1.25rem; flex: 1; font-size: 0.9rem; }
.block-explanation p { margin-bottom: 0.75rem; }
.block-explanation strong { color: var(--text); }
.block-explanation ul { margin-left: 1.25rem; margin-bottom: 0.75rem; }
.block-explanation li { margin-bottom: 0.3rem; }
.diff-section { border-top: 1px solid var(--border); padding: 0; }
.diff-section summary { padding: 0.75rem 1.25rem; cursor: pointer; font-size: 0.85rem; color: var(--text-muted); }
.diff-section summary:hover { color: var(--text); }
pre.diff { margin: 0; padding: 0 1.25rem 1rem; font-family: 'SF Mono', SFMono-Regular, Consolas, monospace; font-size: 0; line-height: 0; overflow-x: auto; }
.diff-line { display: block; font-size: 0.72rem; line-height: 1.45; padding: 1px 0.5rem; white-space: pre; }
.diff-line.add { background: var(--green-bg); color: var(--green); }
.diff-line.remove { background: var(--red-bg); color: var(--red); }
.diff-line.context { color: var(--text-muted); }
.diff-line.ellipsis { color: var(--text-muted); text-align: center; font-style: italic; }
.diff-line .kw { color: var(--syn-keyword); }
.diff-line .str { color: var(--syn-string); }
.diff-line .dec { color: var(--syn-decorator); }
.diff-line .fn { color: var(--syn-func); }
.diff-line .cls { color: var(--syn-class); }
.diff-line .ty { color: var(--syn-type); }
.diff-line .bi { color: var(--syn-builtin); }
.diff-line .num { background: none; width: auto; height: auto; border-radius: 0; display: inline; font-weight: normal; font-size: inherit; color: var(--syn-number); }
.diff-line .slf { color: var(--syn-self); font-style: italic; }
.diff-line .cmt { color: var(--syn-comment); font-style: italic; }
.concerns { border-top: 1px solid var(--border); padding: 1rem 1.25rem; }
.concern-box { padding: 0.75rem 1rem; border-radius: 6px; margin-bottom: 0.5rem; font-size: 0.85rem; }
.concern-box:last-child { margin-bottom: 0; }
.concern-issue { background: rgba(210,153,34,0.1); border-left: 3px solid var(--yellow); }
.concern-ok { background: rgba(63,185,80,0.08); border-left: 3px solid var(--green); }
.summary-table { margin-top: 2rem; }
.summary-table h2 { font-size: 1.2rem; margin-bottom: 1rem; }
table { width: 100%; border-collapse: collapse; background: var(--surface); border: 1px solid var(--border); border-radius: 8px; overflow: hidden; }
th { text-align: left; padding: 0.75rem 1rem; background: rgba(22,27,34,0.8); border-bottom: 1px solid var(--border); font-size: 0.8rem; color: var(--text-muted); text-transform: uppercase; letter-spacing: 0.05em; }
td { padding: 0.75rem 1rem; border-bottom: 1px solid var(--border); font-size: 0.85rem; }
tr:last-child td { border-bottom: none; }
.severity { padding: 0.2rem 0.6rem; border-radius: 12px; font-size: 0.72rem; font-weight: 600; text-transform: uppercase; display: inline-block; }
.severity.design { background: rgba(240,136,62,0.15); color: var(--orange); }
.severity.minor { background: rgba(210,153,34,0.15); color: var(--yellow); }
.severity.bug { background: rgba(248,81,73,0.15); color: var(--red); }
.severity.question { background: rgba(188,140,255,0.15); color: var(--purple); }
.severity.nit { background: rgba(139,148,158,0.15); color: var(--text-muted); }
@media (max-width: 768px) {
.block-body { flex-direction: column; }
.block-files { width: 100%; border-right: none; border-bottom: 1px solid var(--border); }
}
</style>
</head>
<body>
<div class="header">
<h1>PR #4639: TECH-6312 - Asyncio infrastructure for prefork Celery workers</h1>
<div class="meta">
<span>Author: <strong>eric239</strong></span>
<span>Branch: <strong>tech-6312-concurrency-infra-for-use-within-prefork-celery-tasks</strong></span>
<span>Files: <strong>14</strong></span>
<span><span style="color:var(--green)">+429</span> / <span style="color:var(--red)">-15</span></span>
</div>
</div>
<div class="tldr">
<p>This PR introduces a shared asyncio event-loop infrastructure (<code>async_infra</code>) backed by anyio's <code>BlockingPortal</code>, wired into prefork Celery workers via process init/shutdown signals. It includes a new <code>shared-tests</code> library for reusable test fixtures and thorough test coverage. The design is solid with sensible lifecycle management, a fallback path for non-worker contexts, and health-check/restart resilience.</p>
</div>
<!-- Architecture Diagram -->
<div class="diagram-container">
<svg viewBox="0 0 900 400" xmlns="http://www.w3.org/2000/svg" style="max-width:900px;">
<defs>
<marker id="arrow" viewBox="0 0 10 7" refX="10" refY="3.5" markerWidth="10" markerHeight="7" markerUnits="userSpaceOnUse" orient="auto">
<path d="M 0 0 L 10 3.5 L 0 7 z" fill="#8b949e"/>
</marker>
<marker id="arrow-green" viewBox="0 0 10 7" refX="10" refY="3.5" markerWidth="10" markerHeight="7" markerUnits="userSpaceOnUse" orient="auto">
<path d="M 0 0 L 10 3.5 L 0 7 z" fill="#3fb950"/>
</marker>
</defs>
<!-- Celery Worker Process -->
<rect x="30" y="30" width="200" height="60" rx="8" fill="#161b22" stroke="#d29922" stroke-width="2"/>
<text x="130" y="55" fill="#e6edf3" text-anchor="middle" font-size="13" font-weight="600">Celery Worker (prefork)</text>
<text x="130" y="75" fill="#8b949e" text-anchor="middle" font-size="10">api/celery.py</text>
<!-- worker_process_init signal -->
<line x1="230" y1="60" x2="340" y2="60" stroke="#3fb950" stroke-width="1.5" marker-end="url(#arrow-green)"/>
<text x="285" y="52" fill="#3fb950" text-anchor="middle" font-size="9">worker_process_init</text>
<!-- start_async_infra -->
<rect x="345" y="30" width="210" height="60" rx="8" fill="#161b22" stroke="#3fb950" stroke-width="2" onclick="scrollToCard('block-1')" style="cursor:pointer"/>
<text x="450" y="55" fill="#3fb950" text-anchor="middle" font-size="13" font-weight="600">start_async_infra()</text>
<text x="450" y="75" fill="#8b949e" text-anchor="middle" font-size="10">shared_utils.async_infra</text>
<!-- BlockingPortal -->
<line x1="450" y1="90" x2="450" y2="140" stroke="#3fb950" stroke-width="1.5" marker-end="url(#arrow-green)"/>
<rect x="345" y="145" width="210" height="60" rx="8" fill="#161b22" stroke="#3fb950" stroke-width="2"/>
<text x="450" y="170" fill="#3fb950" text-anchor="middle" font-size="13" font-weight="600">BlockingPortal</text>
<text x="450" y="190" fill="#8b949e" text-anchor="middle" font-size="10">anyio daemon thread + asyncio loop</text>
<!-- run_async / run_async_parallel -->
<rect x="345" y="240" width="210" height="60" rx="8" fill="#161b22" stroke="#3fb950" stroke-width="2" onclick="scrollToCard('block-1')" style="cursor:pointer"/>
<text x="450" y="265" fill="#3fb950" text-anchor="middle" font-size="13" font-weight="600">run_async() / parallel()</text>
<text x="450" y="285" fill="#8b949e" text-anchor="middle" font-size="10">submit coros from sync code</text>
<!-- Arrow from run_async to portal -->
<line x1="450" y1="240" x2="450" y2="210" stroke="#8b949e" stroke-width="1.5" marker-end="url(#arrow)"/>
<text x="480" y="228" fill="#8b949e" font-size="9">portal.call()</text>
<!-- Fallback path -->
<rect x="630" y="240" width="210" height="60" rx="8" fill="#161b22" stroke="#58a6ff" stroke-width="2" stroke-dasharray="5,3"/>
<text x="735" y="265" fill="#58a6ff" text-anchor="middle" font-size="13" font-weight="600">asyncio.run() fallback</text>
<text x="735" y="285" fill="#8b949e" text-anchor="middle" font-size="10">when _started is False</text>
<line x1="555" y1="270" x2="625" y2="270" stroke="#58a6ff" stroke-width="1.5" stroke-dasharray="4,3" marker-end="url(#arrow)"/>
<text x="590" y="262" fill="#8b949e" font-size="9">if !started</text>
<!-- get_portal resilience -->
<rect x="630" y="145" width="210" height="60" rx="8" fill="#161b22" stroke="#3fb950" stroke-width="2" onclick="scrollToCard('block-1')" style="cursor:pointer"/>
<text x="735" y="170" fill="#3fb950" text-anchor="middle" font-size="13" font-weight="600">get_portal()</text>
<text x="735" y="190" fill="#8b949e" text-anchor="middle" font-size="10">restart-or-die resilience</text>
<line x1="555" y1="175" x2="625" y2="175" stroke="#3fb950" stroke-width="1.5" marker-end="url(#arrow-green)"/>
<!-- shutdown -->
<rect x="30" y="330" width="200" height="50" rx="8" fill="#161b22" stroke="#d29922" stroke-width="2"/>
<text x="130" y="355" fill="#d29922" text-anchor="middle" font-size="12" font-weight="600">worker_process_shutdown</text>
<line x1="230" y1="355" x2="340" y2="355" stroke="#8b949e" stroke-width="1.5" marker-end="url(#arrow)"/>
<rect x="345" y="330" width="210" height="50" rx="8" fill="#161b22" stroke="#3fb950" stroke-width="2"/>
<text x="450" y="360" fill="#3fb950" text-anchor="middle" font-size="13" font-weight="600">stop_async_infra()</text>
</svg>
</div>
<!-- Block 1 -->
<div class="block-card" id="block-1">
<div class="block-header">
<div class="num">1</div>
<h2>Core async infrastructure (<code>shared_utils.async_infra</code>)</h2>
</div>
<div class="block-body">
<div class="block-files">
<div class="file"><span class="dot dot-added"></span>async_infra.py</div>
<div class="file"><span class="dot dot-modified"></span>shared-utils/pyproject.toml</div>
</div>
<div class="block-explanation">
<p>Provides module-level functions to start/stop a long-lived asyncio event loop on a daemon thread via anyio's <code>BlockingPortal</code>. <code>run_async</code> and <code>run_async_parallel</code> submit coroutines from sync code, falling back to <code>asyncio.run()</code> when the portal has never been initialized.</p>
<p><code>get_portal()</code> implements restart-or-die resilience: if the portal is dead it tries one restart, then calls <code>sys.exit(1)</code>.</p>
<p><strong>Design decisions:</strong></p>
<ul>
<li><code>_started</code> is a one-way latch (never reset). Once initialized, <code>asyncio.run()</code> fallback is permanently bypassed.</li>
<li><code>sys.exit(1)</code> on unrecoverable failure is aggressive but correct for prefork workers.</li>
<li><code>_wrap()</code> closure is necessary because <code>BlockingPortal.call()</code> expects a callable, not a coroutine object.</li>
</ul>
</div>
</div>
<details class="diff-section">
<summary>Show diff: async_infra.py (new file, 147 lines)</summary>
<pre class="diff"><span class="diff-line add">+<span class="kw">import</span> asyncio</span>
<span class="diff-line add">+<span class="kw">import</span> sys</span>
<span class="diff-line add">+<span class="kw">import</span> threading</span>
<span class="diff-line add">+<span class="kw">from</span> collections.abc <span class="kw">import</span> <span class="ty">Coroutine</span></span>
<span class="diff-line add">+<span class="kw">from</span> contextlib <span class="kw">import</span> <span class="ty">AbstractContextManager</span></span>
<span class="diff-line add">+</span>
<span class="diff-line add">+<span class="kw">import</span> anyio</span>
<span class="diff-line add">+<span class="kw">import</span> anyio.from_thread</span>
<span class="diff-line add">+<span class="kw">import</span> structlog</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+<span class="kw">from</span> shared_utils.types <span class="kw">import</span> ANY</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+_logger = structlog.get_logger(__name__)</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+_lock = threading.<span class="cls">Lock</span>()</span>
<span class="diff-line add">+_started = <span class="kw">False</span></span>
<span class="diff-line add">+_cm: <span class="ty">AbstractContextManager</span>[anyio.from_thread.<span class="ty">BlockingPortal</span>] | <span class="kw">None</span> = <span class="kw">None</span></span>
<span class="diff-line add">+_portal: anyio.from_thread.<span class="ty">BlockingPortal</span> | <span class="kw">None</span> = <span class="kw">None</span></span>
<span class="diff-line add">+</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+<span class="kw">class</span> <span class="cls">AsyncInfraError</span>(<span class="bi">Exception</span>):</span>
<span class="diff-line add">+ <span class="str">"""Raised when async infra is used from an incompatible calling context."""</span></span>
<span class="diff-line add">+</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+<span class="kw">def</span> <span class="fn">start_async_infra</span>() -> <span class="kw">None</span>:</span>
<span class="diff-line add">+ <span class="kw">global</span> _cm, _portal, _started</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ <span class="kw">with</span> _lock:</span>
<span class="diff-line add">+ <span class="kw">if</span> _portal <span class="kw">is not</span> <span class="kw">None</span>:</span>
<span class="diff-line add">+ _logger.warning(<span class="str">'Async infra already started, skipping'</span>)</span>
<span class="diff-line add">+ <span class="kw">return</span></span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ cm = anyio.from_thread.start_blocking_portal(backend=<span class="str">'asyncio'</span>)</span>
<span class="diff-line add">+ portal = cm.__enter__()</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ _cm = cm</span>
<span class="diff-line add">+ _portal = portal</span>
<span class="diff-line add">+ _started = <span class="kw">True</span></span>
<span class="diff-line add">+ _logger.info(<span class="str">'Async infra started'</span>)</span>
<span class="diff-line ellipsis">...</span>
<span class="diff-line add">+<span class="kw">def</span> <span class="fn">get_portal</span>() -> anyio.from_thread.<span class="ty">BlockingPortal</span>:</span>
<span class="diff-line add">+ _check_calling_context()</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ portal = _portal</span>
<span class="diff-line add">+ <span class="kw">if</span> portal <span class="kw">is not</span> <span class="kw">None</span> <span class="kw">and</span> _is_portal_healthy(portal, is_final_check=<span class="kw">False</span>):</span>
<span class="diff-line add">+ <span class="kw">return</span> portal</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ _logger.warning(<span class="str">'Async infra portal is dead or missing, attempting restart'</span>)</span>
<span class="diff-line add">+ stop_async_infra()</span>
<span class="diff-line add">+ <span class="kw">try</span>:</span>
<span class="diff-line add">+ start_async_infra()</span>
<span class="diff-line add">+ portal = _portal</span>
<span class="diff-line add">+ <span class="kw">except</span> <span class="bi">Exception</span>:</span>
<span class="diff-line add">+ _logger.exception(<span class="str">'Async infra restart failed'</span>)</span>
<span class="diff-line add">+ portal = <span class="kw">None</span></span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ <span class="kw">if</span> portal <span class="kw">is not</span> <span class="kw">None</span> <span class="kw">and</span> _is_portal_healthy(portal, is_final_check=<span class="kw">True</span>):</span>
<span class="diff-line add">+ <span class="kw">return</span> portal</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ sys.exit(<span class="num">1</span>)</span>
</pre>
</details>
<div class="concerns">
<div class="concern-box concern-issue"><strong>#1 Design:</strong> <code>start_async_infra</code>: if <code>cm.__enter__()</code> raises (line 36), the context manager may have already spawned a thread internally. Since <code>__enter__</code> failed, globals stay <code>None</code> (good), but <code>cm.__exit__</code> is never called, potentially leaking the thread. Consider wrapping in try/except to call <code>cm.__exit__()</code> on failure. <em>(1/2)</em></div>
<div class="concern-box concern-issue"><strong>#2 Design:</strong> <code>ANY</code> TypeVar on <code>run_async</code>/<code>run_async_parallel</code> is not properly bound at the function level. Callers will lose type information. Consider a dedicated <code>T = TypeVar('T')</code> with proper generic signatures. <em>(2/2)</em></div>
<div class="concern-box concern-issue"><strong>#3 Minor:</strong> <code>stop_async_infra</code> docstring should document the intentional <code>_started</code> latch behavior, i.e., that stop does not re-enable the fallback mode. <em>(1/2)</em></div>
</div>
</div>
<!-- Block 2 -->
<div class="block-card" id="block-2">
<div class="block-header">
<div class="num">2</div>
<h2>Celery signal wiring</h2>
</div>
<div class="block-body">
<div class="block-files">
<div class="file"><span class="dot dot-modified"></span>api/celery.py</div>
</div>
<div class="block-explanation">
<p>Hooks <code>start_async_infra</code> into <code>worker_process_init</code> (fires for prefork/solo pools only) and <code>stop_async_infra</code> into <code>worker_process_shutdown</code>.</p>
<p>Also refactors <code>install_usr1_signal</code> to early-return for non-gevent pools, moves the gevent import local (since it may not be installed in prefork environments), and renames <code>log_worker_process_shutdown</code> to <code>on_worker_process_shutdown</code>.</p>
<p><strong>Design decisions:</strong></p>
<ul>
<li>Starting in <code>worker_process_init</code> means each forked worker gets its own event loop thread. Correct: you cannot share threads across fork boundaries.</li>
<li>The gevent import was moved from top-level to local with a noqa suppression. Defensible for an optional dependency.</li>
</ul>
</div>
</div>
<details class="diff-section">
<summary>Show diff: api/celery.py</summary>
<pre class="diff"><span class="diff-line context"> <span class="dec">@worker_ready.connect</span></span>
<span class="diff-line context"> <span class="kw">def</span> <span class="fn">install_usr1_signal</span>(*args: <span class="ty">Any</span>, **kwargs: <span class="ty">Any</span>) -> <span class="kw">None</span>:</span>
<span class="diff-line add">+ logger = structlog.getLogger(__name__)</span>
<span class="diff-line add">+ <span class="kw">if not</span> <span class="bi">any</span>(pair == (<span class="str">'--pool'</span>, <span class="str">'gevent'</span>) <span class="kw">for</span> pair <span class="kw">in</span> <span class="bi">zip</span>(sys.argv[<span class="num">1</span>:], sys.argv[<span class="num">2</span>:], strict=<span class="kw">False</span>)):</span>
<span class="diff-line add">+ logger.info(<span class="str">'Not installing custom USR1 signal handler -- not using "gevent" pool'</span>)</span>
<span class="diff-line add">+ <span class="kw">return</span></span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ <span class="kw">from</span> gevent <span class="kw">import</span> util <span class="kw">as</span> gevent_util <span class="cmt"># noqa: PLC0415 - only needed for gevent workers</span></span>
<span class="diff-line add">+</span>
<span class="diff-line context"> <span class="kw">def</span> <span class="fn">dump_stacktraces</span>(*_args: <span class="ty">Any</span>, **_kwargs: <span class="ty">Any</span>) -> <span class="kw">None</span>:</span>
<span class="diff-line ellipsis">...</span>
<span class="diff-line remove">- logger = structlog.getLogger(__name__)</span>
<span class="diff-line remove">- <span class="kw">if</span> <span class="bi">any</span>(pair == (<span class="str">'--pool'</span>, <span class="str">'gevent'</span>) ...):</span>
<span class="diff-line remove">- logger.info(<span class="str">'Installing custom USR1 signal handler'</span>, pid=...)</span>
<span class="diff-line remove">- signal.signal(signal.SIGUSR1, dump_stacktraces)</span>
<span class="diff-line remove">- <span class="kw">else</span>:</span>
<span class="diff-line remove">- logger.info(<span class="str">'Not installing custom USR1 signal handler...'</span>)</span>
<span class="diff-line add">+ logger.info(<span class="str">'Installing custom USR1 signal handler'</span>, pid=psutil.<span class="cls">Process</span>().pid)</span>
<span class="diff-line add">+ signal.signal(signal.SIGUSR1, dump_stacktraces)</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+<span class="dec">@worker_process_init.connect</span> <span class="cmt"># only fires for prefork/solo pools, not gevent</span></span>
<span class="diff-line add">+<span class="kw">def</span> <span class="fn">on_worker_process_init</span>(**_kwargs: <span class="ty">Any</span>) -> <span class="kw">None</span>:</span>
<span class="diff-line add">+ start_async_infra()</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+</span>
<span class="diff-line context"> <span class="dec">@worker_process_shutdown.connect</span></span>
<span class="diff-line remove">-<span class="kw">def</span> <span class="fn">log_worker_process_shutdown</span>(pid: <span class="ty">int</span>, exitcode: <span class="ty">int</span>, **_kwargs: <span class="ty">Any</span>) -> <span class="kw">None</span>:</span>
<span class="diff-line add">+<span class="kw">def</span> <span class="fn">on_worker_process_shutdown</span>(pid: <span class="ty">int</span>, exitcode: <span class="ty">int</span>, **_kwargs: <span class="ty">Any</span>) -> <span class="kw">None</span>:</span>
<span class="diff-line add">+ stop_async_infra()</span>
<span class="diff-line context"> logger = structlog.getLogger(__name__)</span>
</pre>
</details>
<div class="concerns">
<div class="concern-box concern-ok"><strong>No concerns.</strong> The wiring is clean and the signal scope is correct.</div>
</div>
</div>
<!-- Block 3 -->
<div class="block-card" id="block-3">
<div class="block-header">
<div class="num">3</div>
<h2>New <code>shared-tests</code> library</h2>
</div>
<div class="block-body">
<div class="block-files">
<div class="file"><span class="dot dot-added"></span>libs/shared-tests/ (all)</div>
<div class="file"><span class="dot dot-modified"></span>pyproject.toml</div>
<div class="file"><span class="dot dot-modified"></span>vendor-comms/pyproject.toml</div>
</div>
<div class="block-explanation">
<p>Introduces a new workspace package <code>shared-tests</code> to house reusable test fixtures. Currently contains <code>mock_async_infra</code>, a pytest fixture that patches the portal with synchronous <code>asyncio.run()</code> execution.</p>
<p><strong>Design decisions:</strong></p>
<ul>
<li>Separate library keeps test utilities out of production dependencies.</li>
<li>The fixture patches both <code>_started</code> and <code>get_portal</code> so callers get synchronous execution without a real portal.</li>
</ul>
</div>
</div>
<details class="diff-section">
<summary>Show diff: async_infra_fixtures.py (new fixture)</summary>
<pre class="diff"><span class="diff-line add">+<span class="kw">import</span> asyncio</span>
<span class="diff-line add">+<span class="kw">from</span> collections.abc <span class="kw">import</span> <span class="ty">Generator</span></span>
<span class="diff-line add">+<span class="kw">from</span> unittest.mock <span class="kw">import</span> <span class="cls">MagicMock</span></span>
<span class="diff-line add">+<span class="kw">from</span> unittest.mock <span class="kw">import</span> patch</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+<span class="kw">import</span> pytest</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+<span class="kw">from</span> shared_utils <span class="kw">import</span> async_infra</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+<span class="dec">@pytest.fixture()</span></span>
<span class="diff-line add">+<span class="kw">def</span> <span class="fn">mock_async_infra</span>() -> <span class="ty">Generator</span>[<span class="cls">MagicMock</span>]:</span>
<span class="diff-line add">+ mock_portal = <span class="cls">MagicMock</span>()</span>
<span class="diff-line add">+ mock_portal.call = <span class="kw">lambda</span> fn, *args, **kwargs: asyncio.run(fn(*args, **kwargs))</span>
<span class="diff-line add">+ <span class="kw">with</span> patch.object(async_infra, <span class="str">'_started'</span>, <span class="kw">True</span>), patch.object(async_infra, <span class="str">'get_portal'</span>, return_value=mock_portal):</span>
<span class="diff-line add">+ <span class="kw">yield</span> mock_portal</span>
</pre>
</details>
<div class="concerns">
<div class="concern-box concern-issue"><strong>#4 Design:</strong> <code>shared-tests</code> is not listed in <code>tach.toml</code> <code>source_roots</code>. The other shared libs are all listed. Should it be added for consistency? <em>(1/2)</em></div>
<div class="concern-box concern-issue"><strong>#5 Question:</strong> How will downstream consumers (e.g., the main <code>api</code> test suite) discover the <code>mock_async_infra</code> fixture? Will they need an explicit import in their root conftest? <em>(1/2)</em></div>
</div>
</div>
<!-- Block 4 -->
<div class="block-card" id="block-4">
<div class="block-header">
<div class="num">4</div>
<h2>Tests for async infrastructure</h2>
</div>
<div class="block-body">
<div class="block-files">
<div class="file"><span class="dot dot-added"></span>test_async_infra.py</div>
<div class="file"><span class="dot dot-added"></span>conftest.py</div>
</div>
<div class="block-explanation">
<p>Comprehensive coverage for lifecycle (start/stop/double-start), fallback mode (<code>asyncio.run</code>), portal mode, restart resilience (dead portal triggers restart, unrecoverable state exits process), and calling context validation (rejects async and gevent contexts).</p>
<p><strong>Design decisions:</strong></p>
<ul>
<li>Uses <code>unittest.TestCase</code> per coding standards.</li>
<li><code>TestGetPortalResilience.setUp</code> starts then immediately stops, leaving <code>_started=True</code> and <code>_portal=None</code>. Clean simulation of a crashed portal.</li>
</ul>
</div>
</div>
<details class="diff-section">
<summary>Show diff: test_async_infra.py (139 lines, key excerpts)</summary>
<pre class="diff"><span class="diff-line add">+<span class="kw">class</span> <span class="cls">TestRunAsyncFallback</span>(unittest.<span class="cls">TestCase</span>):</span>
<span class="diff-line add">+ <span class="str">"""Tests run_async / run_async_parallel without portal (asyncio.run fallback)."""</span></span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ <span class="kw">def</span> <span class="fn">setUp</span>(<span class="slf">self</span>) -> <span class="kw">None</span>:</span>
<span class="diff-line add">+ async_infra._started = <span class="kw">False</span></span>
<span class="diff-line add">+ async_infra._portal = <span class="kw">None</span></span>
<span class="diff-line ellipsis">...</span>
<span class="diff-line add">+<span class="kw">class</span> <span class="cls">TestGetPortalResilience</span>(unittest.<span class="cls">TestCase</span>):</span>
<span class="diff-line add">+ <span class="str">"""Tests get_portal restart-or-die behavior when the portal is unhealthy."""</span></span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ <span class="kw">def</span> <span class="fn">setUp</span>(<span class="slf">self</span>) -> <span class="kw">None</span>:</span>
<span class="diff-line add">+ async_infra.start_async_infra()</span>
<span class="diff-line add">+ async_infra.stop_async_infra() <span class="cmt"># leave in dead state for testing</span></span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ <span class="dec">@patch.object</span>(async_infra, <span class="str">'start_async_infra'</span>, wraps=async_infra.start_async_infra)</span>
<span class="diff-line add">+ <span class="kw">def</span> <span class="fn">test_dead_portal_triggers_restart</span>(<span class="slf">self</span>, mock_start):</span>
<span class="diff-line add">+ portal = async_infra.get_portal()</span>
<span class="diff-line add">+ mock_start.assert_called_once()</span>
<span class="diff-line add">+ <span class="slf">self</span>.assertIsNotNone(portal)</span>
<span class="diff-line ellipsis">...</span>
<span class="diff-line add">+<span class="kw">class</span> <span class="cls">TestCheckCallingContext</span>(unittest.<span class="cls">TestCase</span>):</span>
<span class="diff-line add">+ <span class="kw">def</span> <span class="fn">test_rejects_async_context</span>(<span class="slf">self</span>) -> <span class="kw">None</span>:</span>
<span class="diff-line add">+ <span class="kw">async def</span> <span class="fn">_call_from_async</span>() -> <span class="kw">None</span>:</span>
<span class="diff-line add">+ async_infra._check_calling_context()</span>
<span class="diff-line add">+</span>
<span class="diff-line add">+ <span class="kw">with</span> <span class="slf">self</span>.assertRaises(<span class="cls">AsyncInfraError</span>, msg=<span class="str">'async context'</span>):</span>
<span class="diff-line add">+ asyncio.run(_call_from_async())</span>
</pre>
</details>
<div class="concerns">
<div class="concern-box concern-issue"><strong>#6 Minor:</strong> <code>TestRunAsyncFallback.setUp</code> resets <code>_started = False</code> manually but has no <code>tearDown</code>. If test ordering changes, one class could leak state into another. Consider adding <code>stop_async_infra()</code> in <code>tearDown</code>. <em>(2/2)</em></div>
</div>
</div>
<!-- Block 5 -->
<div class="block-card" id="block-5">
<div class="block-header">
<div class="num">5</div>
<h2>Dependency and config changes</h2>
</div>
<div class="block-body">
<div class="block-files">
<div class="file"><span class="dot dot-modified"></span>pyproject.toml</div>
<div class="file"><span class="dot dot-modified"></span>tach.toml</div>
<div class="file"><span class="dot dot-modified"></span>uv.lock</div>
</div>
<div class="block-explanation">
<p>Adds <code>anyio>=4.13</code> as a constraint dependency. Adds <code>shared-tests</code> as workspace member and test dependency. Updates mypy paths. Excludes <code>gevent</code> from tach external checks.</p>
<p>Dependencies are in alphabetical order. No migration files in this PR (expected, pure infrastructure).</p>
</div>
</div>
<div class="concerns">
<div class="concern-box concern-ok"><strong>No concerns.</strong> Clean dependency management following established patterns.</div>
</div>
</div>
<!-- Summary Table -->
<div class="summary-table">
<h2>Summary Action Items</h2>
<table>
<thead>
<tr><th>#</th><th>Severity</th><th>Issue</th></tr>
</thead>
<tbody>
<tr>
<td>1</td>
<td><span class="severity design">Design</span></td>
<td><code>start_async_infra</code>: if <code>cm.__enter__()</code> raises, the context manager is leaked. Consider try/except to call <code>cm.__exit__()</code> on failure. <em>(1/2)</em></td>
</tr>
<tr>
<td>2</td>
<td><span class="severity design">Design</span></td>
<td><code>ANY</code> TypeVar is not properly bound on <code>run_async</code>/<code>run_async_parallel</code>. Consider a dedicated <code>T = TypeVar('T')</code> so callers preserve type information. <em>(2/2)</em></td>
</tr>
<tr>
<td>3</td>
<td><span class="severity minor">Minor</span></td>
<td><code>stop_async_infra</code> docstring should document the intentional <code>_started</code> latch behavior. <em>(1/2)</em></td>
</tr>
<tr>
<td>4</td>
<td><span class="severity design">Design</span></td>
<td><code>shared-tests</code> missing from <code>tach.toml</code> <code>source_roots</code>, unlike the other shared libs. <em>(1/2)</em></td>
</tr>
<tr>
<td>5</td>
<td><span class="severity question">Question</span></td>
<td>How will downstream test suites discover <code>mock_async_infra</code>? Explicit conftest import needed? <em>(1/2)</em></td>
</tr>
<tr>
<td>6</td>
<td><span class="severity minor">Minor</span></td>
<td><code>TestRunAsyncFallback</code> missing <code>tearDown</code> to prevent state leakage across test classes. <em>(2/2)</em></td>
</tr>
</tbody>
</table>
<p style="color: var(--text-muted); font-size: 0.75rem; margin-top: 1rem;">
(N/2) = found by N of 2 independent review agents. Items found by both agents indicate consensus.
</p>
</div>
<script>
function scrollToCard(id) {
document.getElementById(id)?.scrollIntoView({ behavior: 'smooth', block: 'start' });
}
</script>
</body>
</html>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment