bowen31337 Concurrency & Async Pattern Analyzer
Overview
The concurrency-patterns skill inspects the project's Python source files, identifies which concurrency model is in use (anyio, asyncio, threading, or sync), flags anti-patterns that cause hard-to-debug bugs, and generates framework-appropriate rules with ready-to-paste code snippets.
| Capability | Description |
|---|---|
| Framework detection | anyio, asyncio, threading, concurrent.futures, or sync |
| Anti-pattern detection | Blocking calls in async, missing locks, bare thread spawns, deprecated APIs |
| Rule generation | Framework-matched snippets for task groups, locks, pools, sync bridges |
| Severity levels | error, warning, info |
| Output formats | Human-readable table or JSON |
Workflow
Do you want to know which async framework the project uses? -> Framework detection
Do you want to find concurrency bugs or anti-patterns? -> Violation scanning
Do you want a code snippet for a specific pattern? -> Pattern snippets
Do you want a full report (framework + rules + violations)? -> Full analysis
Framework Detection
python skills/concurrency_patterns.py detect --path src/
Output:
Detected framework: anyio
Signals: 12 anyio imports, 5 create_task_group usages, 3 async def
Violation Scanning
# Show only anti-patterns found in the codebase
python skills/concurrency_patterns.py detect --violations-only
Example violations reported:
VIOLATION src/worker.py:42 blocking-call-in-async time.sleep(5) inside async def; use 'await anyio.sleep(5)'
VIOLATION src/api.py:87 blocking-call-in-async requests.get(...) inside async def; use httpx.AsyncClient
WARNING src/cache.py:19 missing-async-lock shared dict mutated in async def without asyncio.Lock / anyio.Lock
Pattern Snippets
# List all available pattern names
python skills/concurrency_patterns.py list-patterns
# Print a ready-to-paste snippet
python skills/concurrency_patterns.py snippet --pattern anyio-task-group
python skills/concurrency_patterns.py snippet --pattern asyncio-lock
python skills/concurrency_patterns.py snippet --pattern thread-pool
python skills/concurrency_patterns.py snippet --pattern run-sync-in-thread
python skills/concurrency_patterns.py snippet --pattern fcntl-file-lock
Full Analysis
# Detect framework, generate rules, and scan for violations
python skills/concurrency_patterns.py detect
# JSON output for programmatic consumption
python skills/concurrency_patterns.py detect --json
Programmatic Use
from pathlib import Path
from skills.concurrency_patterns import ConcurrencyAnalyzer
analyzer = ConcurrencyAnalyzer(root=Path("src/"))
report = analyzer.analyze()
print(report.framework) # "anyio" | "asyncio" | "threading" | "sync"
print(report.stats) # {"async_def_count": 42, "await_count": 87, ...}
for rule in report.rules:
print(rule.name, rule.severity, rule.description)
print(rule.snippet) # ready-to-paste code
for v in report.violations:
print(v.file, v.line, v.rule_name, v.message)
if v.snippet_fix:
print("Fix:", v.snippet_fix)
Generated Rules Reference
anyio rules
| Rule | Severity | Description |
|---|---|---|
anyio-task-group |
info | Use anyio.create_task_group() for structured concurrency |
anyio-lock |
info | Use anyio.Lock() to guard shared mutable state in async code |
anyio-sleep |
error | Replace time.sleep(n) with await anyio.sleep(n) in async functions |
anyio-run-sync |
warning | Use await anyio.to_thread.run_sync(fn) to offload blocking sync calls |
anyio-cancel-scope |
info | Use anyio.CancelScope(deadline=...) for timeout-bounded operations |
asyncio rules
| Rule | Severity | Description |
|---|---|---|
asyncio-task-group |
info | Use asyncio.TaskGroup (3.11+) or asyncio.gather for concurrent tasks |
asyncio-lock |
info | Use asyncio.Lock() to guard shared mutable state |
asyncio-sleep |
error | Replace time.sleep(n) with await asyncio.sleep(n) in async functions |
asyncio-run-in-executor |
warning | Use loop.run_in_executor(None, fn) to offload blocking sync calls |
asyncio-get-running-loop |
warning | Replace deprecated asyncio.get_event_loop() with asyncio.get_running_loop() |
threading rules
| Rule | Severity | Description |
|---|---|---|
thread-lock-context |
info | Use threading.Lock() as a context manager (with lock:) |
thread-pool-sizing |
warning | Size ThreadPoolExecutor(max_workers=...) explicitly; I/O-bound: cpu+4, CPU-bound: cpu |
thread-join-required |
error | Every Thread(...).start() must be paired with .join() or use a pool |
thread-rlock-reentrant |
info | Use threading.RLock() when the same thread may acquire the lock recursively |
sync / multi-process rules
| Rule | Severity | Description |
|---|---|---|
fcntl-file-lock |
info | Use fcntl.LOCK_EX for exclusive cross-process file writes |
atomic-rename |
info | Use tmp.replace(target) (POSIX atomic) for safe file updates |
o-creat-excl-lock |
info | Use `os.O_CREAT |
Anti-Pattern Detection Reference
| Anti-pattern | Detected by | Severity |
|---|---|---|
time.sleep( inside async def |
Regex scan | error |
requests.get/post/put/delete( inside async def |
Regex scan | error |
open( for write inside async def without await |
Regex scan | warning |
asyncio.get_event_loop() (deprecated) |
Regex scan | warning |
Thread(...).start() without .join() |
Heuristic scan | error |
Shared mutable assignment inside async def without lock |
Heuristic scan | warning |
Key Files
| Path | Purpose |
|---|---|
skills/concurrency_patterns.py |
Full implementation — ConcurrencyAnalyzer, PatternRule, Violation, ConcurrencyReport, CLI. |
skills/concurrency-patterns/SKILL.md |
This document — agent routing metadata and usage guide. |
tests/test_concurrency_patterns.py |
pytest test suite for the analyzer. |