ln-653-runtime-performance-auditor

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Runtime Performance Auditor (L3 Worker)

运行时性能审计Worker（L3 Worker）

Specialized worker auditing runtime performance anti-patterns in async and general code.

专门用于审计异步代码和通用代码中运行时性能反模式的Worker。

Purpose & Scope

目标与范围

Worker in ln-650 coordinator pipeline - invoked by ln-650-persistence-performance-auditor
Audit runtime performance (Priority: MEDIUM)
Check async anti-patterns, unnecessary allocations, blocking operations
Return structured findings with severity, location, effort, recommendations
Calculate compliance score (X/10) for Runtime Performance category

ln-650协调器流水线中的Worker - 由ln-650-persistence-performance-auditor调用
审计运行时性能（优先级：MEDIUM）
检查异步反模式、不必要的内存分配、阻塞操作
返回包含严重程度、位置、修复工作量和建议的结构化检测结果
计算运行时性能类别的合规分数（X/10）

Inputs (from Coordinator)

输入（来自协调器）

MANDATORY READ: Load

shared/references/task_delegation_pattern.md#audit-coordinator--worker-contract

for contextStore structure.

Receives

contextStore

with:

tech_stack

best_practices

codebase_root

Domain-aware: Supports

domain_mode

current_domain

必读： 加载

shared/references/task_delegation_pattern.md#audit-coordinator--worker-contract

以了解contextStore结构。

接收包含以下内容的

contextStore

：

tech_stack

、

best_practices

、

codebase_root

。

领域感知： 支持

domain_mode

current_domain

。

Workflow

工作流程

Parse context from contextStore
- Extract tech_stack, best_practices
- Determine scan_path
- Detect async framework: asyncio (Python), Node.js async, Tokio (Rust)
Scan codebase for violations
- Grep patterns scoped to
```
scan_path
```
- For Rules 1, 3, 5: detect
```
async def
```
  blocks first, then check for violations inside them
Collect findings with severity, location, effort, recommendation
Calculate score using penalty algorithm
Return JSON result to coordinator

解析contextStore中的上下文
- 提取tech_stack、best_practices
- 确定扫描路径
- 检测异步框架：asyncio（Python）、Node.js async、Tokio（Rust）
扫描代码库以查找违规项
- 在
```
scan_path
```
  范围内匹配模式
- 对于规则1、3、5：先检测
```
async def
```
  块，再检查其中的违规情况
收集包含严重程度、位置、修复工作量和建议的检测结果
使用惩罚算法计算分数
向协调器返回JSON结果

Audit Rules (Priority: MEDIUM)

审计规则（优先级：MEDIUM）

1. Blocking IO in Async

1. 异步代码中的阻塞IO

What: Synchronous file/network operations inside async functions, blocking event loop

Detection (Python):

Find
```
async def
```
functions

Inside them, grep for blocking calls:

File:

open(

.read_bytes()

.read_text()

.write_bytes()

.write_text()

Path(...).(read|write)

Network:
```
requests.get
```
,
```
requests.post
```
,
```
urllib.request
```
Subprocess:
```
subprocess.run(
```
,
```
subprocess.call(
```

Exclude: calls wrapped in

await asyncio.to_thread(...)

await loop.run_in_executor(...)

Detection (Node.js):

Inside

async function

or arrow async, grep for

fs.readFileSync

fs.writeFileSync

child_process.execSync

Severity:

HIGH: Blocking IO in API request handler (blocks entire event loop)
MEDIUM: Blocking IO in background task/worker

Recommendation: Use

aiofiles

asyncio.to_thread()

, or

loop.run_in_executor()

for file operations; use

httpx.AsyncClient

instead of

requests

Effort: S (wrap in to_thread or switch to async library)

说明： 异步函数内的同步文件/网络操作，会阻塞事件循环

检测方式（Python）：

查找
```
async def
```
函数

在函数内匹配阻塞调用：

文件操作：

open(

.read_bytes()

.read_text()

.write_bytes()

.write_text()

Path(...).(read|write)

网络操作：
```
requests.get
```
,
```
requests.post
```
,
```
urllib.request
```
子进程：
```
subprocess.run(
```
,
```
subprocess.call(
```

排除：被

await asyncio.to_thread(...)

或

await loop.run_in_executor(...)

包裹的调用

检测方式（Node.js）：

在

async function

或异步箭头函数内，匹配

fs.readFileSync

fs.writeFileSync

child_process.execSync

严重程度：

HIGH： API请求处理程序中的阻塞IO（会阻塞整个事件循环）
MEDIUM： 后台任务/Worker中的阻塞IO

建议： 文件操作使用

aiofiles

、

asyncio.to_thread()

或

loop.run_in_executor()

；网络请求使用

httpx.AsyncClient

替代

requests

修复工作量： S（包裹在to_thread中或切换到异步库）

2. Unnecessary List Allocation

2. 不必要的列表分配

What: List comprehension where generator expression suffices

Detection:

```
len([x for x in ...])
```
- allocates list just to count; use
```
sum(1 for ...)
```
```
any([x for x in ...])
```
- allocates list for short-circuit check; use
```
any(x for ...)
```
```
all([x for x in ...])
```
- same pattern; use
```
all(x for ...)
```
```
set([x for x in ...])
```
- use set comprehension
```
{x for x in ...}
```

"".join([x for x in ...])

- use generator directly

"".join(x for x in ...)

Severity:

MEDIUM: Unnecessary allocation in hot path (API handler, loop)
LOW: Unnecessary allocation in infrequent code

Recommendation: Replace

[...]

with generator

(...)

or set comprehension

{...}

Effort: S (syntax change only)

说明： 使用列表推导式，但生成器表达式即可满足需求的情况

检测方式：

```
len([x for x in ...])
```
- 仅为了计数而分配列表；应使用
```
sum(1 for ...)
```
```
any([x for x in ...])
```
- 为短路检查分配列表；应使用
```
any(x for ...)
```
```
all([x for x in ...])
```
- 相同模式；应使用
```
all(x for ...)
```
```
set([x for x in ...])
```
- 应使用集合推导式
```
{x for x in ...}
```

"".join([x for x in ...])

- 直接使用生成器

"".join(x for x in ...)

严重程度：

MEDIUM： 热点路径（API处理程序、循环）中的不必要分配
LOW： 不频繁执行的代码中的不必要分配

建议： 将

[...]

替换为生成器

(...)

或集合推导式

{...}

修复工作量： S（仅需修改语法）

3. Sync Sleep in Async

3. 异步代码中的同步休眠

What:

time.sleep()

inside async function blocks event loop

Detection:

Grep for
```
time\.sleep
```
inside
```
async def
```
blocks

Pattern:

await some_async_call()

...

time.sleep(N)

...

await another_call()

Severity:

HIGH:
```
time.sleep()
```
in async API handler (freezes all concurrent requests)
MEDIUM:
```
time.sleep()
```
in async background task

Recommendation: Replace with

await asyncio.sleep(N)

Effort: S (one-line change)

说明： 异步函数内的

time.sleep()

会阻塞事件循环

检测方式：

在
```
async def
```
块中匹配
```
time\.sleep
```

模式：

await some_async_call()

...

time.sleep(N)

...

await another_call()

严重程度：

HIGH： 异步API处理程序中的
```
time.sleep()
```
（会冻结所有并发请求）
MEDIUM： 异步后台任务中的
```
time.sleep()
```

建议： 替换为

await asyncio.sleep(N)

修复工作量： S（仅需修改一行代码）

4. String Concatenation in Loop

4. 循环中的字符串拼接

What: Building string via

+=

inside loop (O(n^2) for large strings)

Detection:

Pattern: variable
```
result
```
,
```
output
```
,
```
html
```
,
```
text
```
with
```
+=
```
inside
```
for
```
/
```
while
```
loop
Grep for: variable followed by
```
+=
```
containing string operand inside loop body

Severity:

MEDIUM: String concat in loop processing large data (>100 iterations)
LOW: String concat in loop with small iterations (<100)

Recommendation: Use

list.append()

"".join()

, or

io.StringIO

, or f-string with

"".join(generator)

Effort: S (refactor to list + join)

说明： 在循环内通过

+=

构建字符串（处理大字符串时时间复杂度为O(n²)）

检测方式：

模式：变量
```
result
```
,
```
output
```
,
```
html
```
,
```
text
```
在
```
for
```
/
```
while
```
循环内使用
```
+=
```
匹配：循环体中变量后跟包含字符串操作数的
```
+=
```

严重程度：

MEDIUM： 处理大量数据（>100次迭代）的循环中的字符串拼接
LOW： 迭代次数少（<100次）的循环中的字符串拼接

建议： 使用

list.append()

"".join()

，或

io.StringIO

，或结合生成器的f-string +

"".join(generator)

修复工作量： S（重构为列表+join）

5. Missing

to_thread

for CPU-Bound

5. CPU密集型任务缺失

to_thread

What: CPU-intensive synchronous code in async handler without offloading to thread

Detection:

Inside
```
async def
```
, find CPU-intensive operations:
- JSON parsing large files:
```
json.loads(large_data)
```
  ,
```
json.load(file)
```
- Image processing:
```
PIL.Image.open
```
  ,
```
cv2.imread
```
- Crypto:
```
hashlib
```
  ,
```
bcrypt.hashpw
```
- XML/HTML parsing:
```
lxml.etree.parse
```
  ,
```
BeautifulSoup(
```
- Large data transformation without await points
Exclude: operations already wrapped in
```
asyncio.to_thread()
```
or executor

Severity:

MEDIUM: CPU-bound operation in async handler (blocks event loop proportionally to data size)

Recommendation: Wrap in

await asyncio.to_thread(func, *args)

(Python 3.9+) or

loop.run_in_executor(None, func, *args)

Effort: S (wrap in to_thread)

说明： 异步处理程序中的CPU密集型同步代码未卸载到线程

检测方式：

在
```
async def
```
内查找CPU密集型操作：
- 大文件JSON解析：
```
json.loads(large_data)
```
  ,
```
json.load(file)
```
- 图像处理：
```
PIL.Image.open
```
  ,
```
cv2.imread
```
- 加密操作：
```
hashlib
```
  ,
```
bcrypt.hashpw
```
- XML/HTML解析：
```
lxml.etree.parse
```
  ,
```
BeautifulSoup(
```
- 无await点的大数据转换
排除：已被
```
asyncio.to_thread()
```
或执行器包裹的操作

严重程度：

MEDIUM： 异步处理程序中的CPU密集型操作（会根据数据大小成比例阻塞事件循环）

建议： 使用

await asyncio.to_thread(func, *args)

（Python 3.9+）或

loop.run_in_executor(None, func, *args)

包裹

修复工作量： S（包裹在to_thread中）

6. Redundant Data Copies

6. 冗余数据拷贝

What: Unnecessary

.copy()

list()

dict()

when data is only read, not mutated

Detection:

```
data = list(items)
```
where
```
data
```
is only iterated (never modified)
```
config = config_dict.copy()
```
where
```
config
```
is only read
```
result = dict(original)
```
where
```
result
```
is returned without modification

Severity:

LOW: Redundant copy in most contexts (minor memory overhead)
MEDIUM: Redundant copy of large data in hot path

Recommendation: Remove unnecessary copy; pass original if not mutated

Effort: S (remove copy call)

说明： 数据仅用于读取、不修改时，使用不必要的

.copy()

list()

dict()

检测方式：

```
data = list(items)
```
其中
```
data
```
仅被迭代（从未修改）
```
config = config_dict.copy()
```
其中
```
config
```
仅被读取
```
result = dict(original)
```
其中
```
result
```
未修改即被返回

严重程度：

LOW： 大多数场景下的冗余拷贝（内存开销较小）
MEDIUM： 热点路径中的大数据冗余拷贝

建议： 移除不必要的拷贝；若不修改则直接传递原始数据

修复工作量： S（移除拷贝调用）

Scoring Algorithm

评分算法

See

shared/references/audit_scoring.md

for unified formula and score interpretation.

请参阅

shared/references/audit_scoring.md

了解统一公式和分数解读。

Output Format

输出格式

Return JSON to coordinator:

json

{
  "category": "Runtime Performance",
  "score": 7,
  "total_issues": 5,
  "critical": 0,
  "high": 2,
  "medium": 2,
  "low": 1,
  "findings": [
    {
      "severity": "HIGH",
      "location": "app/infrastructure/messaging/job_processor.py:444",
      "issue": "Blocking IO: input_path.read_bytes() inside async function blocks event loop",
      "principle": "Async Best Practices / Non-Blocking IO",
      "recommendation": "Use aiofiles or await asyncio.to_thread(input_path.read_bytes)",
      "effort": "S"
    }
  ]
}

向协调器返回JSON：

json

{
  "category": "Runtime Performance",
  "score": 7,
  "total_issues": 5,
  "critical": 0,
  "high": 2,
  "medium": 2,
  "low": 1,
  "findings": [
    {
      "severity": "HIGH",
      "location": "app/infrastructure/messaging/job_processor.py:444",
      "issue": "Blocking IO: input_path.read_bytes() inside async function blocks event loop",
      "principle": "Async Best Practices / Non-Blocking IO",
      "recommendation": "Use aiofiles or await asyncio.to_thread(input_path.read_bytes)",
      "effort": "S"
    }
  ]
}

Critical Rules

关键规则

Do not auto-fix: Report only
Async context required: Rules 1, 3, 5 apply ONLY inside async functions
Exclude wrappers: Do not flag calls already wrapped in
```
to_thread
```
/
```
run_in_executor
```
Context-aware: Small files (<1KB) read synchronously may be acceptable
Exclude tests: Do not flag test utilities or test fixtures

请勿自动修复： 仅报告问题
需异步上下文： 规则1、3、5仅适用于异步函数内部
排除包裹项： 请勿标记已被
```
to_thread
```
/
```
run_in_executor
```
包裹的调用
上下文感知： 同步读取小文件（<1KB）可能是可接受的
排除测试代码： 请勿标记测试工具或测试夹具

Definition of Done

完成标准

contextStore parsed (tech_stack, async framework)
scan_path determined
Async framework detected (asyncio/Node.js async/Tokio)
All 6 checks completed:
- blocking IO, unnecessary allocations, sync sleep, string concat, CPU-bound, redundant copies
Findings collected with severity, location, effort, recommendation
Score calculated
JSON returned to coordinator

Version: 1.0.0 Last Updated: 2026-02-04

已解析contextStore（tech_stack、异步框架）
已确定scan_path
已检测异步框架（asyncio/Node.js async/Tokio）
已完成全部6项检查：
- 阻塞IO、不必要的分配、同步休眠、字符串拼接、CPU密集型任务、冗余拷贝
已收集包含严重程度、位置、修复工作量和建议的检测结果
已计算分数
已向协调器返回JSON结果

版本： 1.0.0 最后更新： 2026-02-04

ln-653-runtime-performance-auditor

Original

Translation

Runtime Performance Auditor (L3 Worker)

运行时性能审计Worker（L3 Worker）

Purpose & Scope

目标与范围

Inputs (from Coordinator)

输入（来自协调器）

Workflow

工作流程

Audit Rules (Priority: MEDIUM)

审计规则（优先级：MEDIUM）

1. Blocking IO in Async

1. 异步代码中的阻塞IO

2. Unnecessary List Allocation

2. 不必要的列表分配

3. Sync Sleep in Async

3. 异步代码中的同步休眠

4. String Concatenation in Loop

4. 循环中的字符串拼接

5. Missing
`to_thread`
for CPU-Bound

5. CPU密集型任务缺失
`to_thread`

6. Redundant Data Copies

6. 冗余数据拷贝

Scoring Algorithm

评分算法

Output Format

输出格式

Critical Rules

关键规则

Definition of Done

完成标准

ln-653-runtime-performance-auditor

Original

Translation

Runtime Performance Auditor (L3 Worker)

运行时性能审计Worker（L3 Worker）

Purpose & Scope

目标与范围

Inputs (from Coordinator)

输入（来自协调器）

Workflow

工作流程

Audit Rules (Priority: MEDIUM)

审计规则（优先级：MEDIUM）

1. Blocking IO in Async

1. 异步代码中的阻塞IO

2. Unnecessary List Allocation

2. 不必要的列表分配

3. Sync Sleep in Async

3. 异步代码中的同步休眠

4. String Concatenation in Loop

4. 循环中的字符串拼接

5. Missing to_thread for CPU-Bound

5. CPU密集型任务缺失to_thread

6. Redundant Data Copies

6. 冗余数据拷贝

Scoring Algorithm

评分算法

Output Format

输出格式

Critical Rules

关键规则

Definition of Done

完成标准

5. Missing
`to_thread`
for CPU-Bound

5. CPU密集型任务缺失
`to_thread`