From 8d147b2a767fc54689d8c257eff800dbee62bca2 Mon Sep 17 00:00:00 2001 From: SummerOneTwo <89140025+SummerOneTwo@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:11:24 +0800 Subject: [PATCH 1/7] =?UTF-8?q?feat:=20=E9=87=8D=E6=9E=84=E7=AE=97?= =?UTF-8?q?=E6=B3=95=E5=A4=8D=E6=9D=82=E5=BA=A6=E5=88=86=E6=9E=90=E4=BD=93?= =?UTF-8?q?=E7=B3=BB=E4=B8=BA=E5=A4=9A=E8=A7=84=E6=A8=A1=E9=87=87=E6=A0=B7?= =?UTF-8?q?=E4=B8=8E=E7=BB=8F=E9=AA=8C=E5=80=8D=E7=8E=87=E6=8B=9F=E5=90=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...ale-complexity-empirical-fitting-design.md | 205 +++++ ...scale-complexity-empirical-fitting-plan.md | 79 ++ src/autocode_mcp/tools/audit.py | 91 +- src/autocode_mcp/tools/complexity.py | 857 +++++++++--------- src/autocode_mcp/tools/schemas.py | 1 + src/autocode_mcp/tools/solution_audit.py | 62 +- src/autocode_mcp/utils/execution_monitor.py | 311 +++++++ src/autocode_mcp/utils/ratio_analyzer.py | 285 ++++++ src/autocode_mcp/utils/scale_sampler.py | 93 ++ .../test_complexity_empirical_e2e.py | 231 +++++ .../test_empirical_ratio_analyzer.py | 129 +++ tests/test_tools/test_multi_scale_sampler.py | 70 ++ tests/test_tools/test_solution_audit.py | 124 +++ tests/test_utils/test_execution_monitor.py | 89 ++ 14 files changed, 2176 insertions(+), 451 deletions(-) create mode 100644 docs/designs/multi-scale-complexity-empirical-fitting-design.md create mode 100644 docs/plans/multi-scale-complexity-empirical-fitting-plan.md create mode 100644 src/autocode_mcp/utils/execution_monitor.py create mode 100644 src/autocode_mcp/utils/ratio_analyzer.py create mode 100644 src/autocode_mcp/utils/scale_sampler.py create mode 100644 tests/test_integration/test_complexity_empirical_e2e.py create mode 100644 tests/test_tools/test_empirical_ratio_analyzer.py create mode 100644 tests/test_tools/test_multi_scale_sampler.py create mode 100644 tests/test_tools/test_solution_audit.py create mode 100644 tests/test_utils/test_execution_monitor.py diff --git a/docs/designs/multi-scale-complexity-empirical-fitting-design.md b/docs/designs/multi-scale-complexity-empirical-fitting-design.md new file mode 100644 index 0000000..1fe682e --- /dev/null +++ b/docs/designs/multi-scale-complexity-empirical-fitting-design.md @@ -0,0 +1,205 @@ +# 多规模阶梯数据采样与经验复杂度拟合工具链设计 + +## 一、设计目标与问题分析 + +### 1. 现状问题 +在现有工具链实现中,`src/autocode_mcp/tools/complexity.py`、`src/autocode_mcp/tools/solution_audit.py` 与 `src/autocode_mcp/tools/audit.py` 依赖基于正则表达式的代码静态分析(例如通过匹配 `for` 循环和大括号深度推断循环层级,通过搜索关键词识别算法模式)。 + +这种静态分析方法存在固有缺陷: +- 无法感知算法常数:二重循环若内层迭代次数为固定常数,会被误判定为二次时间复杂度。 +- 无法分析均摊复杂度:对于双指针移动、并查集路径压缩、单调队列等经典算法,虽然代码包含嵌套循环,但总时间复杂度为线性,静态规则会产生大量误报。 +- 无法分析递归与剪枝:记忆化搜索或折半搜索的代码结构无法通过大括号层级推导其实际执行步数。 + +### 2. 设计原则 +- **大语言模型负责理论推演与声明**:由大语言模型深入理解算法逻辑,推导理论渐进复杂度并在参数中声明(如 `claimed_complexity`、算法证明、关键瓶颈点)。 +- **工具链负责物理测量与经验拟合**:工具链通过调度数据生成器构造多个阶梯规模的测试数据,实际编译运行程序并采集纯 CPU 耗时与物理内存,计算时间增长倍率并进行对数线性拟合,实现对大语言模型理论声明的实证核验。 +- **工作流时序两阶段协同**:在步骤 4(解法审计阶段)执行理论声明与结构分析,若生成器尚未构建则标记为待实测;在步骤 6(生成器构建)之后以及步骤 10(测试验证)与步骤 11(全量审计)中自动触发多规模实测与经验拟合,解决时序依赖冲突。 +- **接口兼容与客观门禁**:保持 22 个 MCP 工具签名与现有对外接口不变,在门禁中提供包含多规模实测数据的结构化证据,杜绝基于猜测的误报。 + +--- + +## 二、系统架构与处理流程 + +```mermaid +flowchart TD + A["大语言模型 (Host LLM)"] -->|"推导并声明理论复杂度 (claimed_complexity)"| B["solution_audit_std / solution_analyze"] + B --> C{"生成器 files/gen 是否就绪"} + C -->|"尚未构建 (步骤 4)"| D["记录理论声明,标记待实测"] + C -->|"已构建 (步骤 6 之后 / 步骤 10 / 步骤 11)"| E["多规模阶梯采样器 (MultiScaleSampler)"] + E -->|"自适应计算采样阶梯 (5 点采样)"| F["生成器参数调度与数据生成"] + F --> G["动态执行监测器 (DynamicExecutionMonitor)"] + G -->|"运行纯输入输出基准 (Dummy I/O Runner)"| H["测量环境与启动底噪 T0"] + G -->|"并发运行交互题管道或普通解法"| I["采集纯 CPU 耗时 (utime+stime) 与物理内存"] + I --> J["经验倍率拟合分析器 (EmpiricalRatioAnalyzer)"] + J -->|"对数线性回归与动态倍率核验"| K{"实证比对判定"} + K -->|"拟合吻合且无超时"| L["放行并输出结构化实证证据卡片"] + K -->|"拟合失败或发生超时"| M["阻断并向 next_actions 注入修复建议"] +``` + +--- + +## 三、核心模块详细设计 + +### 1. 多规模阶梯数据采样器(MultiScaleSampler) + +根据大语言模型声明的时间复杂度类别与约束规模 $N_{\max}$,自适应计算采样点数值,避免采样点倒挂: + +1. **多项式时间复杂度($N_{\max} \ge 1000$)**: + 采用 5 点阶梯采样,覆盖不同数量级: + - $N_1 = \max(100, \lfloor N_{\max} \times 0.02 \rfloor)$ + - $N_2 = \max(200, \lfloor N_{\max} \times 0.05 \rfloor)$ + - $N_3 = \max(500, \lfloor N_{\max} \times 0.10 \rfloor)$ + - $N_4 = \max(1000, \lfloor N_{\max} \times 0.30 \rfloor)$ + - $N_5 = N_{\max}$ + 严格保证 $N_1 < N_2 < N_3 < N_4 < N_5 \le N_{\max}$。 + +2. **小规模多项式时间复杂度($N_{\max} < 1000$)**: + 采用等比动态比例缩放: + - $N_1 = \max(10, \lfloor N_{\max} \times 0.20 \rfloor)$ + - $N_2 = \lfloor N_{\max} \times 0.40 \rfloor$ + - $N_3 = \lfloor N_{\max} \times 0.60 \rfloor$ + - $N_4 = \lfloor N_{\max} \times 0.80 \rfloor$ + - $N_5 = N_{\max}$ + +3. **指数与阶乘时间复杂度($N_{\max} \le 30$)**: + 采用增量线性阶梯,杜绝指数爆炸: + - $N_1 = \max(4, N_{\max} - 4)$ + - $N_2 = N_{\max} - 3$ + - $N_3 = N_{\max} - 2$ + - $N_4 = N_{\max} - 1$ + - $N_5 = N_{\max}$ + +4. **对数与常数时间复杂度($O(\log N)$ / $O(1)$)**: + 在基准点 $N_1$ 与极限点 $N_5$ 执行物理运行,验证绝对耗时处于极小常数区间。 + +### 2. 动态执行监测器(DynamicExecutionMonitor) + +1. **纯 CPU 时间采集**: + - 在 Linux 操作系统中,调用 `getrusage(RUSAGE_CHILDREN)` 或直接读取 `/proc/[pid]/stat` 获取子进程的 `utime`(用户态时间)与 `stime`(内核态时间),彻底剥离操作系统进程派生、动态库加载以及管道初始化的物理挂钟时间。 + - 在 Windows 操作系统中,在子进程退出后调用 `GetProcessTimes` 函数分别获取目标进程的 `UserTime` 与 `KernelTime` 之和。在评测期间调用 `timeBeginPeriod(1)` 提升定时器分辨率至 1 毫秒,评测完成后调用 `timeEndPeriod(1)` 还原。 + +2. **环境底噪扣除(Dummy I/O Runner)**: + 在执行待测算法前,先行运行仅包含相同输入读取逻辑的空操作基准程序,测量纯输入输出与环境初始化耗时 $T_0$。在计算算法实际耗时时执行 $T_{\text{calc}} = \max(0.1, T_{\text{measured}} - T_0)$,消除小规模测点下的环境底噪压缩,杜绝小常数二次方算法被误判放行。 + +3. **短耗时自适应提升**: + 当基准点纯计算耗时低于 5 毫秒的安全信噪比门限时,自适应向上调整低阶规模参数,确保采样点处于高信噪比区间。 + +4. **进程与资源安全控制**: + - Linux 环境下创建子进程时开启 `start_new_session=True`,发生超时或异常时向进程组发送 `SIGKILL` 信号(`os.killpg`)进行原子化清理。 + - Windows 环境下为子进程关联 Job Object 并配置 `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE` 标志,由操作系统内核执行深层回收。 + - 标准输入输出采用异步非阻塞读取机制,重定向至带大小上限的缓冲区,避免缓冲区填满引发主控与子进程相互挂起。 + +5. **交互题双向管道调度管线**: + 当 `manifest.json` 声明 `interactive: true` 时,并发启动 `files/interactor` 与 `solutions/sol`,通过匿名管道互联两者的标准输入输出,独立采集 `sol` 进程的 CPU 时间,排除交互器自身的计算与调度干扰。 + +### 3. 经验倍率拟合分析器(EmpiricalRatioAnalyzer) + +1. **动态理论倍率计算模型**: + 废除固定的 10 倍倍率表,根据实际采样的具体规模数值 $N_a$ 与 $N_b$($N_a < N_b$),动态计算理论期望倍率: + $$R_{\text{expected}} = \frac{f(N_b)}{f(N_a)}$$ + 实际观测倍率为 $R_{\text{observed}} = \frac{T_{\text{calc}}(N_b)}{T_{\text{calc}}(N_a)}$。 + 容差区间动态调节: + $$R_{\text{observed}} \in \left[ R_{\text{expected}} \times (1 - \delta), R_{\text{expected}} \times (1 + \delta) \right]$$ + 容差系数 $\delta$ 依据基准点耗时自适应设定。 + +2. **对数线性回归拟合模型**: + 通过 5 个阶梯采样点拟合经验幂函数模型: + $$\ln(T_{\text{calc}}(N)) = \alpha \ln N + \beta$$ + 对拟合所得的幂指数 $\alpha$ 进行区间核验: + - 线性复杂度 $O(N)$:$\alpha \in [0.8, 1.3]$ + - 线性对数复杂度 $O(N \log N)$:$\alpha \in [1.0, 1.4]$ + - 二次方复杂度 $O(N^2)$:$\alpha \in [1.7, 2.3]$ + - 三次方复杂度 $O(N^3)$:$\alpha \in [2.6, 3.4]$ + 结合判定系数 $R^2 \ge 0.85$ 评估拟合优度,消除大常数项对增长率的稀释影响。 + +3. **处理器缓存容量跨越保护**: + 实施双区间联合校验。若仅在极限点 $N_5$ 发生微幅跳跃,且极限点绝对耗时处于时限的安全比例内(例如低于时限的 50%),判定为硬件缓存效应并予以放行,记录分析说明。 + +--- + +## 四、接口契约与数据结构规范 + +### 1. 输入参数归一化与轻量化 + +大语言模型调用 `solution_analyze` 与 `solution_audit_std` 时,`claimed_complexity` 为可选参数(`Optional[str] = None`)。 +工具端实施输入归一化解析: +- 自动消除空白字符、转换为小写字符。 +- 自动去除 LaTeX 转义符(例如 `\log` 转换为 `log`,`\cdot` 转换为 `*`)。 +- 规模变量范围与时限自动从 `.autocode/manifest.json` 中提取,无需大语言模型重复传递。 + +### 2. 结构化返回格式规范 + +#### 验证通过时的返回结构(嵌入 `data["empirical_verification"]`): +```json +{ + "empirical_verification": { + "passed": true, + "verdict": "verified", + "claimed_complexity": "O(n log n)", + "fitted_complexity": "O(n log n)", + "fitted_alpha": 1.12, + "r_squared": 0.985, + "samples": [ + {"n": 2000, "cpu_time_ms": 2.4, "memory_mb": 3.1, "status": "ok"}, + {"n": 5000, "cpu_time_ms": 6.8, "memory_mb": 4.2, "status": "ok"}, + {"n": 10000, "cpu_time_ms": 14.5, "memory_mb": 6.0, "status": "ok"}, + {"n": 30000, "cpu_time_ms": 48.2, "memory_mb": 11.5, "status": "ok"}, + {"n": 100000, "cpu_time_ms": 175.0, "memory_mb": 24.8, "status": "ok"} + ], + "growth_ratios": [ + {"from_n": 10000, "to_n": 100000, "scale_factor": 10.0, "observed_ratio": 12.07, "expected_ratio": 12.5} + ], + "max_scale_headroom_ratio": 0.175 + } +} +``` + +#### 验证失败阻断时的返回结构: +```json +{ + "empirical_verification": { + "passed": false, + "verdict": "ratio_mismatch", + "failure_reason": "Measured growth ratio 92.5x significantly exceeds theoretical expectation [10.0x, 15.0x] for O(n log n)", + "claimed_complexity": "O(n log n)", + "fitted_complexity": "O(n^2)", + "fitted_alpha": 2.04, + "r_squared": 0.991, + "samples": [ + {"n": 2000, "cpu_time_ms": 2.1, "memory_mb": 3.0, "status": "ok"}, + {"n": 5000, "cpu_time_ms": 13.5, "memory_mb": 4.1, "status": "ok"}, + {"n": 10000, "cpu_time_ms": 55.2, "memory_mb": 6.0, "status": "ok"}, + {"n": 30000, "cpu_time_ms": 502.1, "memory_mb": 11.2, "status": "ok"}, + {"n": 100000, "cpu_time_ms": null, "memory_mb": null, "status": "timeout"} + ], + "remediation_advice": "The implementation exhibits quadratic growth O(n^2). Please inspect nested loops, optimize algorithm logic to O(n log n), or update claimed_complexity in manifest if quadratic complexity is intended." + } +} +``` + +在 `problem_audit` 中,若经验拟合未通过,向 `blocking_issues` 追加条目,并在 `next_actions` 列表中生成优先级为 `high` 的优化建议动作。 + +--- + +## 五、测试套件详细设计 + +### 1. 单元测试(Unit Tests) +- `tests/test_tools/test_empirical_ratio_analyzer.py`: + - 典型复杂度曲线拟合测试($O(N), O(N \log N), O(N^2), O(N^3)$ 样本)。 + - 动态倍率期望与容差区间计算测试。 + - 测量波动($\pm 15\%$)抗干扰鲁棒性测试。 + - 逆序耗时样本异常识别测试。 + - 环境底噪扣除稳定性测试。 +- `tests/test_tools/test_multi_scale_sampler.py`: + - 采样点数值计算单调性测试(覆盖多项式、小规模、指数级边界)。 + - 生成器调用参数规范格式化测试。 +- `tests/test_utils/test_dynamic_execution_monitor.py`: + - 跨平台纯 CPU 耗时采集精度测试(Linux 与 Windows)。 + - 异常超时进程树原子化清理测试。 + +### 2. 端到端集成测试(Integration Tests) +- `tests/test_integration/test_complexity_empirical_e2e.py`: + - 正确线性标答放行测试(单调队列或双指针)。 + - 小常数二次方算法阻断测试(在 $N=10000$ 下被准确识别并阻断)。 + - 交互题双向管道执行与 CPU 时间采集测试。 + - 多测数据极端分布(大 $T$ 小 $N$ 与小 $T$ 大 $N$)评测。 diff --git a/docs/plans/multi-scale-complexity-empirical-fitting-plan.md b/docs/plans/multi-scale-complexity-empirical-fitting-plan.md new file mode 100644 index 0000000..93d5551 --- /dev/null +++ b/docs/plans/multi-scale-complexity-empirical-fitting-plan.md @@ -0,0 +1,79 @@ +# 多规模阶梯数据采样与经验复杂度拟合执行计划 + +## 一、概述 +本计划依据 [docs/designs/multi-scale-complexity-empirical-fitting-design.md](file:///home/cvm-204/AutoCode/docs/designs/multi-scale-complexity-empirical-fitting-design.md),实施对 AutoCode 复杂度与算法分析体系的全面重构。 +移除原有的脆弱正则表达式静态分析,建立“大语言模型声明理论复杂度 + 底层工具链多规模物理实测与对数线性倍率拟合”的确定性实证架构。 + +--- + +## 二、任务分解与实施步骤 + +### 任务 1:经验倍率拟合分析器(EmpiricalRatioAnalyzer)与数学单元测试 +- **目标文件**: + - 新建 `src/autocode_mcp/utils/ratio_analyzer.py` + - 新建 `tests/test_tools/test_empirical_ratio_analyzer.py` +- **核心逻辑**: + 1. 动态理论期望倍率计算函数:根据任意给定的两测点规模 $N_a, N_b$ 动态求取 $R_{\text{expected}} = f(N_b) / f(N_a)$。 + 2. 对数线性回归拟合模型:计算 $\ln(T_{\text{calc}}) = \alpha \ln N + \beta$ 的幂指数 $\alpha$ 与判定系数 $R^2$。 + 3. 容差检验与复杂度判定函数:比对实测幂指数 $\alpha$ 与观测倍率,输出结构化判定结果(`verified` 或 `ratio_mismatch`)。 +- **验证命令**: + `uv run pytest tests/test_tools/test_empirical_ratio_analyzer.py -v` + +--- + +### 任务 2:自适应多规模阶梯数据采样器(MultiScaleSampler)与单元测试 +- **目标文件**: + - 新建 `src/autocode_mcp/utils/scale_sampler.py` + - 新建 `tests/test_tools/test_multi_scale_sampler.py` +- **核心逻辑**: + 1. 自适应采样点计算:依据 $N_{\max}$ 与时间复杂度类别自适应计算 5 个单调递增采样点。杜绝 $N_{\max} \le 30$ 时的数值倒挂。 + 2. 生成器调用参数规范化:支持位置参数、命名参数与多测参数注入。 + 3. 多测($T$ 组数据)极端数据分布生成逻辑(大 $T$ 小 $N$ 与小 $T$ 大 $N$)。 +- **验证命令**: + `uv run pytest tests/test_tools/test_multi_scale_sampler.py -v` + +--- + +### 任务 3:纯 CPU 耗时采集与环境底噪扣除监控器(DynamicExecutionMonitor) +- **目标文件**: + - 新建 `src/autocode_mcp/utils/execution_monitor.py` + - 新建 `tests/test_utils/test_execution_monitor.py` +- **核心逻辑**: + 1. 跨平台纯 CPU 耗时采集:Linux 环境使用 `getrusage` / `/proc/[pid]/stat`;Windows 环境调用 `GetProcessTimes` 并配合 `timeBeginPeriod(1)`。 + 2. 空操作基准程序(Dummy I/O Runner)测量环境底噪与启动耗时 $T_0$,计算纯算法计算耗时 $T_{\text{calc}} = \max(0.1, T_{\text{measured}} - T_0)$。 + 3. 交互题双向匿名管道并发调度与独立 CPU 耗时统计。 + 4. 进程树生命周期管控:Linux 会话隔离与 `os.killpg`;Windows Job Object 内核清理。 +- **验证命令**: + `uv run pytest tests/test_utils/test_execution_monitor.py -v` + +--- + +### 任务 4:工具层整合与脆弱正则清理 +- **目标文件**: + - 修改 `src/autocode_mcp/tools/complexity.py`:清理 `analyze_loop_complexity` 与 `detect_algorithm_patterns` 中的正则表达式猜测试图;接入 `MultiScaleSampler`、`DynamicExecutionMonitor` 与 `EmpiricalRatioAnalyzer`。 + - 修改 `src/autocode_mcp/tools/solution_audit.py`:在 `solution_audit_std` 与 `solution_audit_brute` 中支持 `claimed_complexity` 归一化输入与多规模经验拟合结果嵌入。 + - 修改 `src/autocode_mcp/tools/audit.py`:在全量审计中将经验拟合指标纳入质量信号,失败时向 `blocking_issues` 与 `next_actions` 追加明确指引。 +- **两阶段时序控制**: + 在解法审计(步骤 4)时若 `files/gen` 未生成,标记为待实测;在生成器构建后及步骤 10 与步骤 11 自动执行完整多规模经验验证。 +- **验证命令**: + `uv run pytest tests/test_tools/test_complexity.py tests/test_tools/test_solution_audit.py -v` + +--- + +### 任务 5:端到端集成测试与小常数二次方算法证伪验证 +- **目标文件**: + - 新建 `tests/test_integration/test_complexity_empirical_e2e.py` +- **核心测试用例**: + 1. 线性标答(单调队列)在 $N=10^5$ 规模下通过验证并输出完整实测证据卡片。 + 2. 声明 $O(N)$ 但实现小常数 $O(N^2)$ 的伪装算法在 $N=10^4$ 规模下被准确识别为二次方增长并就地阻断。 + 3. 交互题双向管道执行与纯 CPU 时间采集准确性测试。 + 4. 生成器参数缺失时的极限单点自适应回退测试。 +- **验证命令**: + `uv run pytest tests/test_integration/test_complexity_empirical_e2e.py -v` + +--- + +### 任务 6:代码规范与全量回归测试 +- **静态代码检查**:`uv run ruff check .` +- **类型系统检查**:`uv run mypy src/` +- **全量测试套件执行**:`uv run pytest tests/ -q` diff --git a/src/autocode_mcp/tools/audit.py b/src/autocode_mcp/tools/audit.py index a78b881..be5b57e 100644 --- a/src/autocode_mcp/tools/audit.py +++ b/src/autocode_mcp/tools/audit.py @@ -1,16 +1,9 @@ -"""High-level problem audit tool. - -This tool aggregates deterministic evidence from the AutoCode problem package. -It does not call an LLM; the LLM-facing difficulty explanation can consume the -returned signals. -""" - from __future__ import annotations import json from datetime import datetime, timezone from pathlib import Path -from typing import Literal +from typing import Any, Literal from pydantic import ValidationError @@ -19,7 +12,12 @@ from ..workflow.guard import signal_satisfied as _guard_signal_satisfied from ..workflow.models import AutoCodeManifest from .base import Tool, ToolResult, input_schema_from_model -from .complexity import analyze_loop_complexity, detect_algorithm_patterns +from .complexity import ( + analyze_loop_complexity, + detect_algorithm_patterns, + extract_claimed_complexity, + run_empirical_verification, +) from .schemas import ProblemAuditInput from .test_verify import ProblemVerifyTestsTool @@ -120,6 +118,37 @@ async def execute( blocking=blocking, next_actions=next_actions, ) + empirical_signal = await self._empirical_complexity_signal(problem_path, manifest) + quality_signals["empirical_complexity"] = empirical_signal + if empirical_signal.get("executed") and not empirical_signal.get("passed"): + evidence = empirical_signal.get("evidence") + failure_reason = ( + str(evidence.get("failure_reason", "empirical complexity verification failed")) + if isinstance(evidence, dict) + else "empirical complexity verification failed" + ) + blocking.append( + { + "gate": "empirical_complexity", + "reason": failure_reason, + } + ) + next_actions.append( + { + "tool_name": "solution_analyze", + "tool": "solution_analyze", + "action": "verify_empirical_complexity", + "recommended_arguments": { + "problem_dir": str(problem_path), + "solution_type": "sol", + }, + "arguments": { + "problem_dir": str(problem_path), + "solution_type": "sol", + }, + "priority": "high", + } + ) statement_consistency = self._statement_consistency(problem_path, manifest, tests_manifest) if statement_consistency["needs_human_review"]: @@ -349,6 +378,42 @@ async def _require_special_artifact_gates( } ) + async def _empirical_complexity_signal( + self, problem_path: Path, manifest: AutoCodeManifest + ) -> dict[str, Any]: + sol_source = self._solution_source(problem_path, "sol") + claimed_complexity = None + if sol_source and sol_source.is_file(): + code = sol_source.read_text(encoding="utf-8", errors="replace") + claimed_complexity = extract_claimed_complexity(code) + + constraints = None + if manifest.constraints: + constraint_numbers = self._constraint_numbers(manifest.constraints) + n_max = max(constraint_numbers) if constraint_numbers else 10000 + constraints = {"n_max": n_max, "time_limit_ms": float(manifest.time_limit_ms)} + + res = await run_empirical_verification( + str(problem_path), + "sol", + claimed_complexity, + constraints, + ) + + if res.get("status") == "pending_generator": + return { + "executed": False, + "passed": True, + "evidence": res, + } + + passed = bool(res.get("passed")) + return { + "executed": True, + "passed": passed, + "evidence": res, + } + def _statement_consistency( self, problem_path: Path, manifest: AutoCodeManifest, tests_manifest: dict ) -> dict[str, object]: @@ -390,6 +455,12 @@ def _difficulty_signals( if pattern_complexity: complexity = self._max_complexity(complexity, pattern_complexity) + empirical_signal = quality_signals.get("empirical_complexity", {}) + if self._signal_satisfied(empirical_signal): + fitted_comp = empirical_signal.get("evidence", {}).get("fitted_complexity") + if fitted_comp: + complexity = str(fitted_comp) + constraint_numbers = self._constraint_numbers(manifest.constraints) n_max = max(constraint_numbers) if constraint_numbers else None wrong_count = sum(1 for s in manifest.solutions if s.role == "wrong") @@ -425,6 +496,8 @@ def _difficulty_signals( confidence += 0.1 if self._signal_satisfied(quality_signals.get("limit_semantics", {})): confidence += 0.1 + if self._signal_satisfied(empirical_signal): + confidence += 0.1 confidence = min(1.0, confidence) reasons = [ diff --git a/src/autocode_mcp/tools/complexity.py b/src/autocode_mcp/tools/complexity.py index d6c96fa..e5e1518 100644 --- a/src/autocode_mcp/tools/complexity.py +++ b/src/autocode_mcp/tools/complexity.py @@ -1,438 +1,421 @@ -""" -Complexity 分析工具 - 分析解法复杂度。 - -基于代码静态分析估算时间/空间复杂度,并推荐测试参数。 -""" - -from __future__ import annotations - -import os -import re -from typing import Any - -from .base import Tool, ToolResult, input_schema_from_model -from .mixins import resolve_source -from .schemas import SolutionAnalyzeInput - - -class ComplexityLevel: - """复杂度等级。""" - - CONSTANT = "O(1)" - LOG_N = "O(log n)" - LINEAR = "O(n)" - N_LOG_N = "O(n log n)" - QUADRATIC = "O(n^2)" - CUBIC = "O(n^3)" - EXPONENTIAL = "O(2^n)" - FACTORIAL = "O(n!)" - - -# 复杂度到推荐 n_max 的映射 -COMPLEXITY_TO_N_MAX = { - ComplexityLevel.CONSTANT: 10**9, - ComplexityLevel.LOG_N: 10**9, - ComplexityLevel.LINEAR: 10**7, - ComplexityLevel.N_LOG_N: 10**6, - ComplexityLevel.QUADRATIC: 5000, - ComplexityLevel.CUBIC: 500, - ComplexityLevel.EXPONENTIAL: 20, - ComplexityLevel.FACTORIAL: 12, -} - -# 复杂度到推荐时间限制的映射(毫秒) -COMPLEXITY_TO_TIME_LIMIT = { - ComplexityLevel.CONSTANT: 1000, - ComplexityLevel.LOG_N: 1000, - ComplexityLevel.LINEAR: 1000, - ComplexityLevel.N_LOG_N: 2000, - ComplexityLevel.QUADRATIC: 3000, - ComplexityLevel.CUBIC: 5000, - ComplexityLevel.EXPONENTIAL: 10000, - ComplexityLevel.FACTORIAL: 10000, -} - - -def analyze_loop_complexity(code: str) -> str: - """分析循环复杂度。 - - Args: - code: C++ 源代码 - - Returns: - 估算的复杂度字符串 - """ - # 循环模式 - loop_patterns = [ - r"\bfor\s*\(", - r"\bwhile\s*\(", - r"\bfor\s+\w+.*:", # range-based for - ] - - max_nesting = 0 - brace_depth = 0 - saw_loop = False - - lines = code.split("\n") - for line in lines: - # 移除单行注释 - if "//" in line: - line = line[: line.index("//")] - - # 检测当前行是否有循环 - has_loop = any(re.search(p, line) for p in loop_patterns) - - # 如果有循环,记录当前深度 - # brace_depth 表示当前所在的大括号层级 - # 循环嵌套数 = 当前大括号层级 - if has_loop: - saw_loop = True - max_nesting = max(max_nesting, brace_depth) - - # 处理大括号 - for char in line: - if char == "{": - brace_depth += 1 - elif char == "}": - brace_depth = max(0, brace_depth - 1) - - # 根据嵌套层数估算复杂度 - if not saw_loop: - return ComplexityLevel.CONSTANT - if max_nesting == 0: - return ComplexityLevel.LINEAR - elif max_nesting == 1: - return ComplexityLevel.LINEAR - elif max_nesting == 2: - return ComplexityLevel.QUADRATIC - elif max_nesting == 3: - return ComplexityLevel.CUBIC - else: - return ComplexityLevel.EXPONENTIAL - - -def detect_algorithm_patterns(code: str) -> tuple[str | None, list[str]]: - """检测常见算法模式。 - - Args: - code: C++ 源代码 - - Returns: - (复杂度或 None, 检测到的模式列表) - 如果没有检测到模式,返回 (None, []) - """ - patterns = [] - complexity = None # 默认不返回复杂度 - - # 二分查找 - if re.search(r"\bbinary_search\b|\blower_bound\b|\bupper_bound\b", code): - patterns.append("binary_search") - complexity = ComplexityLevel.N_LOG_N - - # 排序 - if re.search(r"\bsort\b|\bstable_sort\b|\bpartial_sort\b", code): - patterns.append("sorting") - complexity = ComplexityLevel.N_LOG_N - - # 归并与分治 - if re.search(r"\bmerge(_sort|_count)?\b|\bdivide_and_conquer\b", code, re.IGNORECASE): - patterns.append("divide_and_conquer") - complexity = ComplexityLevel.N_LOG_N - - # 树状结构 - if re.search(r"\b(fenwick|bit|segtree|segment_tree)\b", code, re.IGNORECASE): - patterns.append("tree_data_structure") - complexity = ComplexityLevel.N_LOG_N - - # 图算法 - BFS/DFS - if re.search(r"\bbfs\b|\bdfs\b|queue<|stack<", code): - patterns.append("graph_traversal") - complexity = ComplexityLevel.LINEAR - - # 动态规划 - if re.search(r"dp\[|memo\[|memoization", code): - patterns.append("dynamic_programming") - # DP 复杂度取决于状态数和转移 - complexity = ComplexityLevel.QUADRATIC - - # 哈希表 - if re.search(r"unordered_map|unordered_set|hash_map", code): - patterns.append("hash_table") - # 如果主要操作是哈希,可能更优 - - # 递归 - if re.search(r"\breturn\s+\w+\s*\([^)]*\)", code) and re.search( - r"\b\w+\s*\([^)]*\)\s*{", code - ): - # 简单的递归检测 - patterns.append("recursion") - - # 位运算 - if re.search(r"1\s*<<\s*\w+|bitmask|bitset", code): - patterns.append("bitmask") - complexity = ComplexityLevel.EXPONENTIAL - - return complexity, patterns - - -def extract_claimed_complexity(code: str) -> str | None: - match = re.search(r"O\([^)]*\)", code) - if not match: - return None - return match.group(0) - - -def build_risk_notes( - estimated: str, constraints: dict[str, Any] | None, warnings: list[str] -) -> list[str]: - notes = list(warnings) - if estimated in {ComplexityLevel.QUADRATIC, ComplexityLevel.CUBIC}: - notes.append("高复杂度实现对 n 上限敏感,建议强化 type=4(TLE) 对拍。") - if estimated in {ComplexityLevel.EXPONENTIAL, ComplexityLevel.FACTORIAL}: - notes.append("指数级/阶乘级复杂度通常不适合作为标准解,请核对题面约束。") - if constraints and constraints.get("n_max", 0) >= 10**6 and estimated not in { - ComplexityLevel.LINEAR, - ComplexityLevel.N_LOG_N, - ComplexityLevel.LOG_N, - ComplexityLevel.CONSTANT, - }: +from __future__ import annotations + +import asyncio +import os +import re +from typing import Any + +from ..utils.execution_monitor import DynamicExecutionMonitor +from ..utils.platform import get_exe_extension +from ..utils.ratio_analyzer import EmpiricalRatioAnalyzer, normalize_complexity_expression +from ..utils.scale_sampler import MultiScaleSampler +from .base import Tool, ToolResult, input_schema_from_model +from .mixins import resolve_source +from .schemas import SolutionAnalyzeInput + + +class ComplexityLevel: + CONSTANT = "O(1)" + LOG_N = "O(log n)" + LINEAR = "O(n)" + N_LOG_N = "O(n log n)" + N_SQRT_N = "O(n sqrt n)" + QUADRATIC = "O(n^2)" + CUBIC = "O(n^3)" + EXPONENTIAL = "O(2^n)" + FACTORIAL = "O(n!)" + + +COMPLEXITY_TO_N_MAX = { + ComplexityLevel.CONSTANT: 10**9, + ComplexityLevel.LOG_N: 10**9, + ComplexityLevel.LINEAR: 10**7, + ComplexityLevel.N_LOG_N: 10**6, + ComplexityLevel.QUADRATIC: 5000, + ComplexityLevel.CUBIC: 500, + ComplexityLevel.EXPONENTIAL: 20, + ComplexityLevel.FACTORIAL: 12, +} + +COMPLEXITY_TO_TIME_LIMIT = { + ComplexityLevel.CONSTANT: 1000, + ComplexityLevel.LOG_N: 1000, + ComplexityLevel.LINEAR: 1000, + ComplexityLevel.N_LOG_N: 2000, + ComplexityLevel.QUADRATIC: 3000, + ComplexityLevel.CUBIC: 5000, + ComplexityLevel.EXPONENTIAL: 10000, + ComplexityLevel.FACTORIAL: 10000, +} + + +def analyze_loop_complexity(code: str) -> str: + loop_patterns = [ + r"\bfor\s*\(", + r"\bwhile\s*\(", + r"\bfor\s+\w+.*:", + ] + + max_nesting = 0 + brace_depth = 0 + saw_loop = False + + lines = code.split("\n") + for line in lines: + if "//" in line: + line = line[: line.index("//")] + + has_loop = any(re.search(p, line) for p in loop_patterns) + if has_loop: + saw_loop = True + max_nesting = max(max_nesting, brace_depth) + + for char in line: + if char == "{": + brace_depth += 1 + elif char == "}": + brace_depth = max(0, brace_depth - 1) + + if not saw_loop: + return ComplexityLevel.CONSTANT + if max_nesting <= 1: + return ComplexityLevel.LINEAR + elif max_nesting == 2: + return ComplexityLevel.QUADRATIC + elif max_nesting == 3: + return ComplexityLevel.CUBIC + return ComplexityLevel.EXPONENTIAL + + +def detect_algorithm_patterns(code: str) -> tuple[str | None, list[str]]: + patterns: list[str] = [] + complexity: str | None = None + + if re.search(r"\bbinary_search\b|\blower_bound\b|\bupper_bound\b", code): + patterns.append("binary_search") + complexity = ComplexityLevel.N_LOG_N + + if re.search(r"\bsort\b|\bstable_sort\b|\bpartial_sort\b", code): + patterns.append("sorting") + complexity = ComplexityLevel.N_LOG_N + + if re.search(r"\bmerge(_sort|_count)?\b|\bdivide_and_conquer\b", code, re.IGNORECASE): + patterns.append("divide_and_conquer") + complexity = ComplexityLevel.N_LOG_N + + if re.search(r"\b(fenwick|bit|segtree|segment_tree)\b", code, re.IGNORECASE): + patterns.append("tree_data_structure") + complexity = ComplexityLevel.N_LOG_N + + if re.search(r"\bbfs\b|\bdfs\b|queue<|stack<", code): + patterns.append("graph_traversal") + complexity = ComplexityLevel.LINEAR + + if re.search(r"dp\[|memo\[|memoization", code): + patterns.append("dynamic_programming") + complexity = ComplexityLevel.QUADRATIC + + if re.search(r"unordered_map|unordered_set|hash_map", code): + patterns.append("hash_table") + + if re.search(r"\breturn\s+\w+\s*\([^)]*\)", code) and re.search( + r"\b\w+\s*\([^)]*\)\s*{", code + ): + patterns.append("recursion") + + if re.search(r"1\s*<<\s*\w+|bitmask|bitset", code): + patterns.append("bitmask") + complexity = ComplexityLevel.EXPONENTIAL + + return complexity, patterns + + +def extract_claimed_complexity(code: str) -> str | None: + match = re.search(r"O\([^)]*\)", code) + if not match: + return None + return match.group(0) + + +def build_risk_notes( + estimated: str, constraints: dict[str, Any] | None, warnings: list[str] +) -> list[str]: + notes = list(warnings) + if estimated in {ComplexityLevel.QUADRATIC, ComplexityLevel.CUBIC}: + notes.append("高复杂度实现对 n 上限敏感,建议强化极限对拍。") + if estimated in {ComplexityLevel.EXPONENTIAL, ComplexityLevel.FACTORIAL}: + notes.append("指数级复杂度通常不适合作为标准解,请核对题面约束。") + if constraints and constraints.get("n_max", 0) >= 10**6 and estimated not in { + ComplexityLevel.LINEAR, + ComplexityLevel.N_LOG_N, + ComplexityLevel.LOG_N, + ComplexityLevel.CONSTANT, + }: notes.append("n_max 较大,复杂度分析过高,存在超时风险,建议参考。") - return notes - - -def estimate_memory_usage(code: str) -> tuple[str, int]: - """估算内存使用。 - - Args: - code: C++ 源代码 - - Returns: - (空间复杂度描述, 估算的内存 MB) - """ - # 检测大数组 - array_patterns = [ - r"(\w+)\s*\[(\d+)\]", # int arr[1000] - r"vector<\w+>\s+(\w+)\s*\((\d+)\)", # vector v(1000) - r"array<\w+,\s*(\d+)>", # array - ] - - total_elements = 0 - for pattern in array_patterns: - matches = re.findall(pattern, code) - for match in matches: - try: - # 获取数字部分 - if isinstance(match, tuple): - size = int(match[-1]) - else: - size = int(match) - total_elements += size - except (ValueError, IndexError): - pass - - # 估算内存(假设每个元素 4 字节) - memory_bytes = total_elements * 4 - memory_mb = max(1, memory_bytes // (1024 * 1024)) - - if total_elements == 0: - return "O(1)", 64 - elif total_elements < 10000: - return "O(n)", memory_mb - elif total_elements < 1000000: - return "O(n)", memory_mb - else: - return "O(n) - large", memory_mb - - -class SolutionAnalyzeTool(Tool): - """分析解法复杂度。""" - - @property - def name(self) -> str: - return "solution_analyze" - - @property - def description(self) -> str: - return """分析 C++ 解法代码的时间/空间复杂度。 - - 基于静态分析估算: - - 时间复杂度(循环嵌套、算法模式) - - 空间复杂度(数组、容器大小) - - 推荐的测试参数 - - 前置条件: - 1. 已有解法代码(可以是未编译的源码) - - 建议下一步: - - 根据推荐的 n_max 调整测试数据生成参数 - - 根据推荐的 time_limit 设置题目时间限制 - """ - - @property - def input_schema(self) -> dict: - return input_schema_from_model(SolutionAnalyzeInput) - - async def execute( - self, - code: str | None = None, - problem_dir: str | None = None, - solution_type: str = "sol", - source_path: str | None = None, - constraints: dict | None = None, - ) -> ToolResult: - """执行复杂度分析。""" - if solution_type not in {"sol", "brute"}: - return ToolResult.fail("solution_type must be 'sol' or 'brute'") - if code is None and source_path is None and not problem_dir: - return ToolResult.fail("Either 'code', 'source_path', or 'problem_dir' must be provided") - - resolved, err = resolve_source( - problem_dir or ".", - code, - source_path, - default_source_path=os.path.join("solutions", f"{solution_type}.cpp"), - ) - if err is not None: - return err - assert resolved is not None - code = resolved.code - - # 1. 分析循环复杂度 - loop_complexity = analyze_loop_complexity(code) - - # 2. 检测算法模式 - pattern_complexity, patterns = detect_algorithm_patterns(code) - - # 3. 选择复杂度估计 - # 如果检测到算法模式,取两者中较大的(更保守的估计) - if pattern_complexity is not None: - complexity_order = [ - ComplexityLevel.CONSTANT, - ComplexityLevel.LOG_N, - ComplexityLevel.LINEAR, - ComplexityLevel.N_LOG_N, - ComplexityLevel.QUADRATIC, - ComplexityLevel.CUBIC, - ComplexityLevel.EXPONENTIAL, - ComplexityLevel.FACTORIAL, - ] - - loop_idx = complexity_order.index(loop_complexity) - pattern_idx = complexity_order.index(pattern_complexity) - - # 取较大的复杂度(更保守) - final_complexity = ( - pattern_complexity if pattern_idx > loop_idx else loop_complexity - ) - else: - final_complexity = loop_complexity - - # 4. 估算内存 - space_complexity, memory_mb = estimate_memory_usage(code) - - # 5. 生成推荐参数 - recommended_n_max = COMPLEXITY_TO_N_MAX.get(final_complexity, 10000) - recommended_time_ms = COMPLEXITY_TO_TIME_LIMIT.get(final_complexity, 1000) - claimed_complexity = extract_claimed_complexity(code) - - # 如果有题目约束,验证是否合理 - warnings = [] - if constraints: - if constraints.get("n_max"): - if constraints["n_max"] > recommended_n_max: - warnings.append( - f"Warning: n_max={constraints['n_max']} may cause TLE " - f"for {final_complexity} algorithm. Recommended: {recommended_n_max}" - ) - if constraints.get("time_limit_ms"): - if constraints["time_limit_ms"] < recommended_time_ms: - warnings.append( - f"Warning: time_limit={constraints['time_limit_ms']}ms may be too tight " - f"for {final_complexity} algorithm. Recommended: {recommended_time_ms}ms" - ) - - risk_notes = build_risk_notes(final_complexity, constraints, warnings) - suggested_test_configs = self._generate_test_configs(recommended_n_max, constraints) - stress_profiles = self._recommended_stress_profiles( - final_complexity=final_complexity, - recommended_n_max=recommended_n_max, - constraints=constraints, - ) - - return ToolResult.ok( - claimed_complexity=claimed_complexity, - estimated_complexity=final_complexity, - worst_case_complexity=final_complexity, - average_case_complexity=final_complexity, - time_complexity=final_complexity, - space_complexity=space_complexity, - memory_estimate={"space_complexity": space_complexity, "estimated_memory_mb": memory_mb}, - estimated_memory_mb=memory_mb, - detected_patterns=patterns, - recommended_n_max=recommended_n_max, - recommended_time_limit_ms=recommended_time_ms, - warnings=warnings, - risk_notes=risk_notes, - suggested_test_configs=suggested_test_configs, - recommended_stress_params=stress_profiles, - message=f"Analyzed complexity: {final_complexity}", - ) - - def _generate_test_configs( - self, n_max: int, constraints: dict | None - ) -> list[dict]: - """生成推荐的测试配置。 - - Args: - n_max: 推荐的 n 最大值 - constraints: 题目约束 - - Returns: - 测试配置列表 - """ - # 使用约束中的 n_max 或推荐值 - actual_n_max = constraints.get("n_max", n_max) if constraints else n_max - - configs = [ - # 边界情况 - {"type": "1", "n_min": 1, "n_max": 1, "t_min": 1, "t_max": 1}, - {"type": "1", "n_min": 1, "n_max": 10, "t_min": 1, "t_max": 1}, - # 随机数据 - {"type": "2", "n_min": 10, "n_max": actual_n_max // 10, "t_min": 1, "t_max": 1}, - {"type": "2", "n_min": actual_n_max // 10, "n_max": actual_n_max // 2, "t_min": 1, "t_max": 1}, - # 极限数据 - {"type": "3", "n_min": actual_n_max // 2, "n_max": actual_n_max, "t_min": 1, "t_max": 1}, - {"type": "3", "n_min": actual_n_max, "n_max": actual_n_max, "t_min": 1, "t_max": 1}, - ] - - return configs - - def _recommended_stress_profiles( - self, - final_complexity: str, - recommended_n_max: int, - constraints: dict | None, - ) -> list[dict]: - n_cap = constraints.get("n_max", recommended_n_max) if constraints else recommended_n_max - brute_n = min(max(20, n_cap // 50), 2000) - trials = 300 if final_complexity in {ComplexityLevel.QUADRATIC, ComplexityLevel.CUBIC} else 1000 - return [ - { - "name": "tiny_exhaustive", - "trials": min(200, trials), - "types": ["1"], - "generator_args": {"type": "1", "n_min": 1, "n_max": 8, "t_min": 1, "t_max": 1}, - }, - { - "name": "random_small", - "trials": trials, - "types": ["2"], - "generator_args": {"type": "2", "n_min": 1, "n_max": brute_n, "t_min": 1, "t_max": 1}, - }, - { - "name": "edge_small", - "trials": max(100, trials // 3), - "types": ["3", "4"], - "generator_args": {"type": "3", "n_min": max(1, brute_n // 2), "n_max": brute_n, "t_min": 1, "t_max": 1}, - }, - ] + return notes + + +def estimate_memory_usage(code: str) -> tuple[str, int]: + array_patterns = [ + r"(\w+)\s*\[(\d+)\]", + r"vector<\w+>\s+(\w+)\s*\((\d+)\)", + r"array<\w+,\s*(\d+)>", + ] + + total_elements = 0 + for pattern in array_patterns: + matches = re.findall(pattern, code) + for match in matches: + try: + if isinstance(match, tuple): + size = int(match[-1]) + else: + size = int(match) + total_elements += size + except (ValueError, IndexError): + pass + + memory_bytes = total_elements * 4 + memory_mb = max(1, memory_bytes // (1024 * 1024)) + + if total_elements == 0: + return "O(1)", 64 + elif total_elements < 10000: + return "O(n)", memory_mb + elif total_elements < 1000000: + return "O(n)", memory_mb + return "O(n) - large", memory_mb + + +async def run_empirical_verification( + problem_dir: str, + solution_type: str, + claimed_complexity: str | None, + constraints: dict[str, Any] | None, +) -> dict[str, Any]: + exe_ext = get_exe_extension() + gen_path = os.path.join(problem_dir, "files", f"gen{exe_ext}") + sol_path = os.path.join(problem_dir, "solutions", f"{solution_type}{exe_ext}") + + if not os.path.isfile(gen_path) or not os.path.isfile(sol_path): + return { + "status": "pending_generator", + "message": "files/gen or binary not built yet, empirical verification will execute after generator_build", + } + + actual_n_max = 10000 + time_limit_ms = 2000.0 + if constraints: + actual_n_max = int(constraints.get("n_max") or 10000) + time_limit_ms = float(constraints.get("time_limit_ms") or 2000.0) + + effective_complexity = claimed_complexity or "O(n)" + scale_points = MultiScaleSampler.compute_scale_points(actual_n_max, effective_complexity) + + empirical_dir = os.path.join(problem_dir, ".autocode", "empirical_tests") + os.makedirs(empirical_dir, exist_ok=True) + + baseline_overhead = await DynamicExecutionMonitor.measure_baseline_overhead(cwd=problem_dir) + + samples: list[dict[str, Any]] = [] + for idx, pt in enumerate(scale_points): + in_file = os.path.join(empirical_dir, f"scale_{pt}.in") + gen_cmd = MultiScaleSampler.format_generator_command(gen_path, pt, 42 + idx) + try: + await asyncio.to_thread( + MultiScaleSampler.generate_scale_input_file, + gen_cmd, + in_file, + 5.0, + ) + except Exception as e: + return { + "status": "generator_error", + "message": f"Failed generating scale point {pt}: {e}", + } + + res = await DynamicExecutionMonitor.run_monitored_process( + [sol_path], + in_file, + time_limit_ms=time_limit_ms, + cwd=problem_dir, + baseline_overhead_ms=baseline_overhead, + ) + samples.append({ + "n": pt, + "cpu_time_ms": res.get("cpu_time_ms"), + "memory_mb": res.get("memory_mb"), + "status": res.get("status"), + }) + if res.get("status") in ("timeout", "mle"): + break + + return EmpiricalRatioAnalyzer.verify_complexity( + effective_complexity, + samples, + time_limit_ms=time_limit_ms, + ) + + +class SolutionAnalyzeTool(Tool): + @property + def name(self) -> str: + return "solution_analyze" + + @property + def description(self) -> str: + return "分析 C++ 解法代码的时间与空间复杂度,结合经验数据拟合输出实证分析证据。" + + @property + def input_schema(self) -> dict: + return input_schema_from_model(SolutionAnalyzeInput) + + async def execute( + self, + code: str | None = None, + problem_dir: str | None = None, + solution_type: str = "sol", + source_path: str | None = None, + constraints: dict | None = None, + claimed_complexity: str | None = None, + ) -> ToolResult: + if solution_type not in {"sol", "brute"}: + return ToolResult.fail("solution_type must be 'sol' or 'brute'") + if code is None and source_path is None and not problem_dir: + return ToolResult.fail("Either 'code', 'source_path', or 'problem_dir' must be provided") + + resolved, err = resolve_source( + problem_dir or ".", + code, + source_path, + default_source_path=os.path.join("solutions", f"{solution_type}.cpp"), + ) + if err is not None: + return err + assert resolved is not None + code = resolved.code + + loop_complexity = analyze_loop_complexity(code) + pattern_complexity, patterns = detect_algorithm_patterns(code) + + if pattern_complexity is not None: + complexity_order = [ + ComplexityLevel.CONSTANT, + ComplexityLevel.LOG_N, + ComplexityLevel.LINEAR, + ComplexityLevel.N_LOG_N, + ComplexityLevel.QUADRATIC, + ComplexityLevel.CUBIC, + ComplexityLevel.EXPONENTIAL, + ComplexityLevel.FACTORIAL, + ] + loop_idx = complexity_order.index(loop_complexity) + pattern_idx = complexity_order.index(pattern_complexity) + final_complexity = pattern_complexity if pattern_idx > loop_idx else loop_complexity + else: + final_complexity = loop_complexity + + if claimed_complexity: + norm_claimed = normalize_complexity_expression(claimed_complexity) + final_complexity = norm_claimed + + space_complexity, memory_mb = estimate_memory_usage(code) + + recommended_n_max = COMPLEXITY_TO_N_MAX.get(final_complexity, 10000) + recommended_time_ms = COMPLEXITY_TO_TIME_LIMIT.get(final_complexity, 1000) + detected_claimed = extract_claimed_complexity(code) + + warnings: list[str] = [] + if constraints: + if constraints.get("n_max"): + if constraints["n_max"] > recommended_n_max: + warnings.append( + f"Warning: n_max={constraints['n_max']} may cause TLE " + f"for {final_complexity} algorithm. Recommended: {recommended_n_max}" + ) + if constraints.get("time_limit_ms"): + if constraints["time_limit_ms"] < recommended_time_ms: + warnings.append( + f"Warning: time_limit={constraints['time_limit_ms']}ms may be too tight " + f"for {final_complexity} algorithm. Recommended: {recommended_time_ms}ms" + ) + + risk_notes = build_risk_notes(final_complexity, constraints, warnings) + suggested_test_configs = self._generate_test_configs(recommended_n_max, constraints) + stress_profiles = self._recommended_stress_profiles( + final_complexity=final_complexity, + recommended_n_max=recommended_n_max, + constraints=constraints, + ) + + empirical_verification: dict[str, Any] = {"status": "skipped"} + if problem_dir: + empirical_verification = await run_empirical_verification( + problem_dir, + solution_type, + claimed_complexity or detected_claimed or final_complexity, + constraints, + ) + + return ToolResult.ok( + claimed_complexity=claimed_complexity or detected_claimed, + estimated_complexity=final_complexity, + worst_case_complexity=final_complexity, + average_case_complexity=final_complexity, + time_complexity=final_complexity, + space_complexity=space_complexity, + memory_estimate={"space_complexity": space_complexity, "estimated_memory_mb": memory_mb}, + estimated_memory_mb=memory_mb, + detected_patterns=patterns, + recommended_n_max=recommended_n_max, + recommended_time_limit_ms=recommended_time_ms, + warnings=warnings, + risk_notes=risk_notes, + suggested_test_configs=suggested_test_configs, + recommended_stress_params=stress_profiles, + empirical_verification=empirical_verification, + message=f"Analyzed complexity: {final_complexity}", + ) + + def _generate_test_configs( + self, n_max: int, constraints: dict | None + ) -> list[dict]: + actual_n_max = constraints.get("n_max", n_max) if constraints else n_max + return [ + {"type": "1", "n_min": 1, "n_max": 1, "t_min": 1, "t_max": 1}, + {"type": "1", "n_min": 1, "n_max": 10, "t_min": 1, "t_max": 1}, + {"type": "2", "n_min": 10, "n_max": actual_n_max // 10, "t_min": 1, "t_max": 1}, + {"type": "2", "n_min": actual_n_max // 10, "n_max": actual_n_max // 2, "t_min": 1, "t_max": 1}, + {"type": "3", "n_min": actual_n_max // 2, "n_max": actual_n_max, "t_min": 1, "t_max": 1}, + {"type": "3", "n_min": actual_n_max, "n_max": actual_n_max, "t_min": 1, "t_max": 1}, + ] + + def _recommended_stress_profiles( + self, + final_complexity: str, + recommended_n_max: int, + constraints: dict | None, + ) -> list[dict]: + n_cap = constraints.get("n_max", recommended_n_max) if constraints else recommended_n_max + brute_n = min(max(20, n_cap // 50), 2000) + trials = 300 if final_complexity in {ComplexityLevel.QUADRATIC, ComplexityLevel.CUBIC} else 1000 + return [ + { + "name": "tiny_exhaustive", + "trials": min(200, trials), + "types": ["1"], + "generator_args": {"type": "1", "n_min": 1, "n_max": 8, "t_min": 1, "t_max": 1}, + }, + { + "name": "random_small", + "trials": trials, + "types": ["2"], + "generator_args": {"type": "2", "n_min": 1, "n_max": brute_n, "t_min": 1, "t_max": 1}, + }, + { + "name": "edge_small", + "trials": max(100, trials // 3), + "types": ["3", "4"], + "generator_args": {"type": "3", "n_min": max(1, brute_n // 2), "n_max": brute_n, "t_min": 1, "t_max": 1}, + }, + ] diff --git a/src/autocode_mcp/tools/schemas.py b/src/autocode_mcp/tools/schemas.py index a6158bd..797ffeb 100644 --- a/src/autocode_mcp/tools/schemas.py +++ b/src/autocode_mcp/tools/schemas.py @@ -180,6 +180,7 @@ class SolutionAnalyzeInput(BaseModel): solution_type: Literal["sol", "brute"] = "sol" source_path: str | None = None constraints: AnalyzeConstraints | None = None + claimed_complexity: str | None = None # --------------------------------------------------------------------------- # diff --git a/src/autocode_mcp/tools/solution_audit.py b/src/autocode_mcp/tools/solution_audit.py index 3e6ee5f..7f01452 100644 --- a/src/autocode_mcp/tools/solution_audit.py +++ b/src/autocode_mcp/tools/solution_audit.py @@ -1,13 +1,14 @@ -""" -Solution 审计工具:审核标准解与暴力解的可行性与复杂度假设。 -""" - from __future__ import annotations import os +from ..utils.ratio_analyzer import normalize_complexity_expression from .base import Tool, ToolResult, input_schema_from_model -from .complexity import ComplexityLevel, analyze_loop_complexity +from .complexity import ( + ComplexityLevel, + analyze_loop_complexity, + run_empirical_verification, +) from .mixins import resolve_source from .schemas import SolutionAuditBruteInput, SolutionAuditStdInput @@ -46,9 +47,14 @@ async def execute( return err assert resolved is not None code = resolved.code + + if claimed_complexity: + claimed_complexity = normalize_complexity_expression(claimed_complexity) + estimated = analyze_loop_complexity(code) findings: list[dict] = [] passed = True + if claimed_complexity and claimed_complexity != estimated: findings.append( { @@ -57,6 +63,7 @@ async def execute( "message": f"claimed={claimed_complexity}, estimated={estimated}", } ) + if constraints and constraints.get("n_max", 0) >= 10**6 and estimated in { ComplexityLevel.QUADRATIC, ComplexityLevel.CUBIC, @@ -70,12 +77,42 @@ async def execute( "message": "n_max 较大,静态分析估算的标准解复杂度过高,存在 TLE 风险,建议参考。", } ) + + empirical_verification: dict = {"status": "skipped"} + if problem_dir: + empirical_verification = await run_empirical_verification( + problem_dir, + "sol", + claimed_complexity, + constraints, + ) + if empirical_verification.get("status") == "pending_generator": + pass + elif empirical_verification.get("status") == "generator_error": + findings.append( + { + "severity": "warning", + "type": "generator_error", + "message": str(empirical_verification.get("message", "生成器运行异常")), + } + ) + elif not empirical_verification.get("passed", True): + passed = False + findings.append( + { + "severity": "error", + "type": "empirical_ratio_mismatch", + "message": str(empirical_verification.get("failure_reason", "经验复杂度拟合未通过")), + } + ) + return ToolResult.ok( passed=passed, estimated_complexity=estimated, claimed_complexity=claimed_complexity, findings=findings, evidence={"has_sort": "sort(" in code, "has_nested_loops": code.count("for (") >= 2}, + empirical_verification=empirical_verification, ) @@ -113,6 +150,10 @@ async def execute( return err assert resolved is not None code = resolved.code + + if std_complexity: + std_complexity = normalize_complexity_expression(std_complexity) + brute_complexity = analyze_loop_complexity(code) findings: list[dict] = [] if std_complexity and brute_complexity == std_complexity: @@ -129,6 +170,16 @@ async def execute( elif brute_complexity == ComplexityLevel.QUADRATIC: recommended_n_max = 80 recommended_trials = 1200 if recommended_n_max >= 100 else 600 + + empirical_verification: dict = {"status": "skipped"} + if problem_dir: + empirical_verification = await run_empirical_verification( + problem_dir, + "brute", + brute_complexity, + constraints, + ) + return ToolResult.ok( passed=True, brute_complexity=brute_complexity, @@ -140,4 +191,5 @@ async def execute( "types": ["1", "2", "3"], "timeout": 30, }, + empirical_verification=empirical_verification, ) diff --git a/src/autocode_mcp/utils/execution_monitor.py b/src/autocode_mcp/utils/execution_monitor.py new file mode 100644 index 0000000..62df0aa --- /dev/null +++ b/src/autocode_mcp/utils/execution_monitor.py @@ -0,0 +1,311 @@ +import asyncio +import os +import time +from typing import Any + +import psutil + +from .process import terminate_pid_tree + + +class DynamicExecutionMonitor: + @classmethod + def get_cpu_times_ms(cls, proc: psutil.Process) -> float: + try: + times = proc.cpu_times() + return (times.user + times.system) * 1000.0 + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0.0 + + @classmethod + def get_memory_mb(cls, proc: psutil.Process) -> float: + try: + mem = proc.memory_info() + return mem.rss / (1024.0 * 1024.0) + except (psutil.NoSuchProcess, psutil.AccessDenied): + return 0.0 + + @classmethod + async def measure_baseline_overhead(cls, cwd: str | None = None) -> float: + try: + cmd = ["/bin/true"] if os.path.exists("/bin/true") else ["true"] + t0 = time.perf_counter() + subproc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + cwd=cwd, + ) + await subproc.wait() + return max(0.0, (time.perf_counter() - t0) * 1000.0) + except Exception: + return 1.5 + + @classmethod + async def run_monitored_process( + cls, + cmd: list[str], + input_file_path: str, + time_limit_ms: float = 2000.0, + memory_limit_mb: float = 512.0, + cwd: str | None = None, + baseline_overhead_ms: float = 0.0, + ) -> dict[str, Any]: + timeout_sec = max(0.1, time_limit_ms / 1000.0) + max_output_bytes = 10 * 1024 * 1024 # 10MB 输出上限,防止管道填满挂起 + + wall_start = time.perf_counter() + peak_memory_mb = 0.0 + accumulated_cpu_ms = 0.0 + + with open(input_file_path, "rb") as in_f: + subproc = await asyncio.create_subprocess_exec( + *cmd, + stdin=in_f, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + start_new_session=True, + ) + + pid = subproc.pid + status = "ok" + stdout_bytes = b"" + stderr_bytes = b"" + + try: + ps_proc = psutil.Process(pid) + except (psutil.NoSuchProcess, psutil.AccessDenied): + ps_proc = None + + async def _sample_resources() -> None: + nonlocal peak_memory_mb, accumulated_cpu_ms + while subproc.returncode is None: + if ps_proc: + try: + with ps_proc.oneshot(): + mem = cls.get_memory_mb(ps_proc) + if mem > peak_memory_mb: + peak_memory_mb = mem + cpu = cls.get_cpu_times_ms(ps_proc) + if cpu > accumulated_cpu_ms: + accumulated_cpu_ms = cpu + if peak_memory_mb > memory_limit_mb: + break + except (psutil.NoSuchProcess, psutil.AccessDenied): + break + await asyncio.sleep(0.005) + + sampler_task = asyncio.create_task(_sample_resources()) + + try: + stdout_data, stderr_data = await asyncio.wait_for( + subproc.communicate(), + timeout=timeout_sec, + ) + stdout_bytes = stdout_data[:max_output_bytes] + stderr_bytes = stderr_data[:max_output_bytes] + except asyncio.TimeoutError: + status = "timeout" + await terminate_pid_tree(pid) + try: + await asyncio.wait_for(subproc.wait(), timeout=1.0) + except Exception: + pass + finally: + sampler_task.cancel() + try: + await sampler_task + except asyncio.CancelledError: + pass + + wall_elapsed_ms = (time.perf_counter() - wall_start) * 1000.0 + + if status != "timeout": + if peak_memory_mb > memory_limit_mb: + status = "mle" + await terminate_pid_tree(pid) + try: + await asyncio.wait_for(subproc.wait(), timeout=1.0) + except Exception: + pass + elif subproc.returncode != 0: + status = "runtime_error" + + calc_wall_ms = max(0.1, wall_elapsed_ms - baseline_overhead_ms) + final_cpu_ms = ( + accumulated_cpu_ms + if (accumulated_cpu_ms and accumulated_cpu_ms > 0.5) + else round(calc_wall_ms, 2) + ) if status == "ok" else None + + return { + "status": status, + "returncode": subproc.returncode, + "cpu_time_ms": final_cpu_ms, + "wall_time_ms": round(wall_elapsed_ms, 2), + "memory_mb": round(peak_memory_mb, 2), + "stdout": stdout_bytes.decode("utf-8", errors="replace"), + "stderr": stderr_bytes.decode("utf-8", errors="replace"), + } + + @classmethod + async def run_interactive_pipeline( + cls, + interactor_cmd: list[str], + solution_cmd: list[str], + input_file_path: str, + time_limit_ms: float = 2000.0, + memory_limit_mb: float = 512.0, + cwd: str | None = None, + ) -> dict[str, Any]: + timeout_sec = max(0.1, time_limit_ms / 1000.0) + + with open(input_file_path, "rb") as in_f: + input_bytes = in_f.read() + + interactor_proc = await asyncio.create_subprocess_exec( + *interactor_cmd, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + start_new_session=True, + ) + + solution_proc = await asyncio.create_subprocess_exec( + *solution_cmd, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + start_new_session=True, + ) + + sol_pid = solution_proc.pid + int_pid = interactor_proc.pid + + try: + sol_ps = psutil.Process(sol_pid) + except (psutil.NoSuchProcess, psutil.AccessDenied): + sol_ps = None + + peak_memory_mb = 0.0 + accumulated_cpu_ms = 0.0 + + async def _sample_sol() -> None: + nonlocal peak_memory_mb, accumulated_cpu_ms + while solution_proc.returncode is None: + if sol_ps: + try: + with sol_ps.oneshot(): + mem = cls.get_memory_mb(sol_ps) + if mem > peak_memory_mb: + peak_memory_mb = mem + accumulated_cpu_ms = cls.get_cpu_times_ms(sol_ps) + if peak_memory_mb > memory_limit_mb: + break + except (psutil.NoSuchProcess, psutil.AccessDenied): + break + await asyncio.sleep(0.02) + + sampler_task = asyncio.create_task(_sample_sol()) + + # 启动管道泵送数据 + async def _pump_stream(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + try: + while True: + chunk = await reader.read(4096) + if not chunk: + break + writer.write(chunk) + await writer.drain() + except (BrokenPipeError, ConnectionResetError, asyncio.CancelledError): + pass + finally: + try: + writer.close() + except Exception: + pass + + async def _drain_stream(reader: asyncio.StreamReader | None) -> None: + if not reader: + return + try: + while True: + chunk = await reader.read(4096) + if not chunk: + break + except (BrokenPipeError, ConnectionResetError, asyncio.CancelledError): + pass + + async def _feed_input() -> None: + if interactor_proc.stdin and input_bytes: + try: + interactor_proc.stdin.write(input_bytes) + await interactor_proc.stdin.drain() + except (BrokenPipeError, ConnectionResetError, asyncio.CancelledError): + pass + finally: + try: + interactor_proc.stdin.close() + except Exception: + pass + + pump_tasks = [] + pump_tasks.append(asyncio.create_task(_feed_input())) + + if interactor_proc.stdout and solution_proc.stdin: + pump_tasks.append(asyncio.create_task(_pump_stream(interactor_proc.stdout, solution_proc.stdin))) + if solution_proc.stdout and interactor_proc.stdin: + pump_tasks.append(asyncio.create_task(_pump_stream(solution_proc.stdout, interactor_proc.stdin))) + + # 异步排空 stderr,防止管道填满阻塞 + if interactor_proc.stderr: + pump_tasks.append(asyncio.create_task(_drain_stream(interactor_proc.stderr))) + if solution_proc.stderr: + pump_tasks.append(asyncio.create_task(_drain_stream(solution_proc.stderr))) + + status = "ok" + try: + await asyncio.wait_for( + asyncio.gather(solution_proc.wait(), interactor_proc.wait()), + timeout=timeout_sec, + ) + except asyncio.TimeoutError: + status = "timeout" + await terminate_pid_tree(sol_pid) + await terminate_pid_tree(int_pid) + try: + await asyncio.wait_for( + asyncio.gather(solution_proc.wait(), interactor_proc.wait()), + timeout=1.0, + ) + except Exception: + pass + finally: + sampler_task.cancel() + for t in pump_tasks: + t.cancel() + try: + await sampler_task + except asyncio.CancelledError: + pass + + if status != "timeout": + if peak_memory_mb > memory_limit_mb: + status = "mle" + await terminate_pid_tree(sol_pid) + await terminate_pid_tree(int_pid) + elif solution_proc.returncode != 0: + status = "runtime_error" + elif interactor_proc.returncode != 0: + status = "interactor_error" + + return { + "status": status, + "solution_returncode": solution_proc.returncode, + "interactor_returncode": interactor_proc.returncode, + "cpu_time_ms": max(accumulated_cpu_ms, 0.1) if status == "ok" else None, + "memory_mb": round(peak_memory_mb, 2), + } diff --git a/src/autocode_mcp/utils/ratio_analyzer.py b/src/autocode_mcp/utils/ratio_analyzer.py new file mode 100644 index 0000000..45991e2 --- /dev/null +++ b/src/autocode_mcp/utils/ratio_analyzer.py @@ -0,0 +1,285 @@ +import math +import re +from typing import Any + + +def normalize_complexity_expression(expr: str | None) -> str: + if not expr or not isinstance(expr, str): + return "O(n)" + cleaned = expr.strip().lower() + cleaned = cleaned.replace("\\log", "log").replace("\\cdot", "*").replace("\\times", "*") + cleaned = re.sub(r"\s+", "", cleaned) + match = re.search(r"o\((.+)\)", cleaned) + if match: + inner = match.group(1) + else: + inner = cleaned + inner = inner.replace("*", "") + inner = inner.replace("log(n)", "logn").replace("sqrt(n)", "sqrtn") + inner = inner.replace("**", "^") + + # 映射为标准包含空格形式 + mapping = { + "1": "O(1)", + "logn": "O(log n)", + "n": "O(n)", + "nlogn": "O(n log n)", + "nsqrtn": "O(n sqrt n)", + "n^2": "O(n^2)", + "n^3": "O(n^3)", + "2^n": "O(2^n)", + "n!": "O(n!)", + } + return mapping.get(inner, f"O({inner})") + + +class EmpiricalRatioAnalyzer: + COMPLEXITY_ALPHA_RANGES: dict[str, tuple[float, float]] = { + "O(1)": (-0.2, 0.35), + "O(log n)": (-0.05, 0.65), + "O(logn)": (-0.05, 0.65), + "O(n)": (0.35, 1.35), + "O(n log n)": (0.55, 1.45), + "O(nlogn)": (0.55, 1.45), + "O(n sqrt n)": (1.15, 1.75), + "O(nsqrtn)": (1.15, 1.75), + "O(n^2)": (1.65, 2.35), + "O(n^3)": (2.50, 3.45), + "O(2^n)": (3.50, 100.0), + "O(n!)": (3.50, 100.0), + } + + @classmethod + def evaluate_complexity_function(cls, normalized_expr: str, n: float) -> float: + if n <= 1: + n = 1.0001 + if normalized_expr in ("O(1)",): + return 1.0 + if normalized_expr in ("O(log n)", "O(logn)"): + return math.log2(max(2.0, n)) + if normalized_expr in ("O(n)",): + return n + if normalized_expr in ("O(n log n)", "O(nlogn)"): + return n * math.log2(max(2.0, n)) + if normalized_expr in ("O(n sqrt n)", "O(nsqrtn)"): + return n * math.sqrt(n) + if normalized_expr in ("O(n^2)",): + return n * n + if normalized_expr in ("O(n^3)",): + return n * n * n + if normalized_expr in ("O(2^n)",): + return math.pow(2.0, min(n, 60.0)) + if normalized_expr in ("O(n!)",): + return float(math.factorial(min(int(n), 20))) + return n + + @classmethod + def calculate_expected_ratio(cls, claimed_complexity: str, n_from: int, n_to: int) -> float: + norm_expr = normalize_complexity_expression(claimed_complexity) + val_from = cls.evaluate_complexity_function(norm_expr, float(n_from)) + val_to = cls.evaluate_complexity_function(norm_expr, float(n_to)) + if val_from <= 0: + return 1.0 + return val_to / val_from + + @classmethod + def fit_log_linear(cls, samples: list[dict[str, Any]]) -> tuple[float, float]: + valid_points: list[tuple[float, float]] = [] + for s in samples: + try: + n_val = float(s.get("n", 0) or 0) + t_val = float(s.get("cpu_time_ms", 0.0) or 0.0) + except (ValueError, TypeError): + continue + if n_val > 1 and t_val > 0.01 and math.isfinite(n_val) and math.isfinite(t_val): + valid_points.append((math.log(n_val), math.log(t_val))) + + if len(valid_points) < 2: + return 1.0, 0.0 + + n_count = len(valid_points) + sum_x = sum(p[0] for p in valid_points) + sum_y = sum(p[1] for p in valid_points) + sum_xx = sum(p[0] * p[0] for p in valid_points) + sum_xy = sum(p[0] * p[1] for p in valid_points) + + denom = n_count * sum_xx - sum_x * sum_x + if abs(denom) < 1e-9: + return 1.0, 0.0 + + alpha = (n_count * sum_xy - sum_x * sum_y) / denom + beta = (sum_y - alpha * sum_x) / n_count + + mean_y = sum_y / n_count + ss_tot = sum((p[1] - mean_y) ** 2 for p in valid_points) + ss_res = sum((p[1] - (alpha * p[0] + beta)) ** 2 for p in valid_points) + + if ss_tot < 1e-9: + r_squared = 1.0 + else: + r_squared = max(0.0, 1.0 - (ss_res / ss_tot)) + + return alpha, r_squared + + @classmethod + def infer_complexity_category(cls, alpha: float) -> str: + if alpha < 0.35: + return "O(1)" + if alpha < 0.65: + return "O(log n)" + if alpha < 1.35: + return "O(n)" + if alpha < 1.65: + return "O(n log n)" + if alpha < 1.85: + return "O(n sqrt n)" + if alpha < 2.50: + return "O(n^2)" + if alpha < 3.50: + return "O(n^3)" + return "O(2^n)" + + @classmethod + def verify_complexity( + cls, + claimed_complexity: str, + samples: list[dict[str, Any]], + time_limit_ms: float = 2000.0, + ) -> dict[str, Any]: + norm_claimed = normalize_complexity_expression(claimed_complexity) + valid_samples = [ + s for s in samples + if s.get("status") == "ok" + and s.get("cpu_time_ms") is not None + and math.isfinite(float(s.get("cpu_time_ms", 0.0) or 0.0)) + ] + + if not valid_samples: + return { + "passed": False, + "verdict": "no_valid_samples", + "failure_reason": "No valid execution samples collected", + "claimed_complexity": norm_claimed, + "fitted_complexity": "unknown", + "fitted_alpha": None, + "r_squared": None, + "samples": samples, + "growth_ratios": [], + "remediation_advice": "Check if generator and solution execute properly without immediate crash.", + } + + last_sample = samples[-1] + if last_sample.get("status") == "timeout": + return { + "passed": False, + "verdict": "timeout_at_extreme", + "failure_reason": f"Execution timed out at extreme test point (N={last_sample.get('n')})", + "claimed_complexity": norm_claimed, + "fitted_complexity": "timeout", + "fitted_alpha": None, + "r_squared": None, + "samples": samples, + "growth_ratios": [], + "remediation_advice": f"Algorithm execution timed out on N={last_sample.get('n')}. Optimize time complexity or reduce constant factors.", + } + + if len(valid_samples) < 2: + return { + "passed": False, + "verdict": "insufficient_samples", + "failure_reason": "Insufficient valid execution samples collected (minimum 2 required)", + "claimed_complexity": norm_claimed, + "fitted_complexity": "unknown", + "fitted_alpha": None, + "r_squared": None, + "samples": samples, + "growth_ratios": [], + "remediation_advice": "Ensure generator produces inputs and solution executes without early crash.", + } + + alpha, r_squared = cls.fit_log_linear(valid_samples) + fitted_cat = cls.infer_complexity_category(alpha) + + expected_range = cls.COMPLEXITY_ALPHA_RANGES.get(norm_claimed, (0.65, 1.35)) + is_alpha_match = expected_range[0] <= alpha <= expected_range[1] + + # 若后段高信噪比测点(后 3 点)呈现显著高阶增长,采用后段拟合结果以防前段底噪平线掩盖真实复杂度 + if len(valid_samples) >= 4: + tail_samples = valid_samples[-3:] + tail_alpha, tail_r2 = cls.fit_log_linear(tail_samples) + if tail_alpha > expected_range[1] + 0.10: + is_alpha_match = False + alpha = tail_alpha + fitted_cat = cls.infer_complexity_category(tail_alpha) + r_squared = tail_r2 + + growth_ratios: list[dict[str, Any]] = [] + for i in range(len(valid_samples) - 1): + s_from = valid_samples[i] + s_to = valid_samples[i + 1] + n_from = int(s_from["n"]) + n_to = int(s_to["n"]) + t_from = float(s_from["cpu_time_ms"]) + t_to = float(s_to["cpu_time_ms"]) + + observed_ratio = (t_to / t_from) if t_from > 0.01 else 1.0 + expected_ratio = cls.calculate_expected_ratio(norm_claimed, n_from, n_to) + + growth_ratios.append( + { + "from_n": n_from, + "to_n": n_to, + "scale_factor": round(float(n_to) / float(n_from), 2) if n_from > 0 else 1.0, + "observed_ratio": round(observed_ratio, 2), + "expected_ratio": round(expected_ratio, 2), + } + ) + + max_sample = valid_samples[-1] + max_time = float(max_sample["cpu_time_ms"]) + max_n = int(max_sample.get("n", 0)) + headroom_ratio = max_time / time_limit_ms if time_limit_ms > 0 else 1.0 + + if not is_alpha_match: + # 仅在规模足够大(跨越硬件缓存)且耗时余量充足时允许缓存跳跃容差 + if alpha > expected_range[1] and len(valid_samples) >= 3 and headroom_ratio < 0.40 and max_n >= 50000: + sub_samples = valid_samples[:-1] + sub_alpha, sub_r2 = cls.fit_log_linear(sub_samples) + if expected_range[0] <= sub_alpha <= expected_range[1]: + return { + "passed": True, + "verdict": "verified_with_cache_jump", + "claimed_complexity": norm_claimed, + "fitted_complexity": norm_claimed, + "fitted_alpha": round(alpha, 2), + "r_squared": round(r_squared, 3), + "samples": samples, + "growth_ratios": growth_ratios, + "max_scale_headroom_ratio": round(headroom_ratio, 3), + "cache_jump_note": "Slight ratio jump observed at maximum scale likely due to L1/L2 cache capacity transition, well within time limit.", + } + + return { + "passed": False, + "verdict": "ratio_mismatch", + "failure_reason": f"Measured complexity exponent alpha={alpha:.2f} (fitted as {fitted_cat}) outside claimed {norm_claimed} range [{expected_range[0]}, {expected_range[1]}]", + "claimed_complexity": norm_claimed, + "fitted_complexity": fitted_cat, + "fitted_alpha": round(alpha, 2), + "r_squared": round(r_squared, 3), + "samples": samples, + "growth_ratios": growth_ratios, + "remediation_advice": f"The algorithm exhibits {fitted_cat} growth rather than claimed {norm_claimed}. Inspect nested loops, optimize algorithm logic, or update claimed_complexity.", + } + + return { + "passed": True, + "verdict": "verified", + "claimed_complexity": norm_claimed, + "fitted_complexity": fitted_cat, + "fitted_alpha": round(alpha, 2), + "r_squared": round(r_squared, 3), + "samples": samples, + "growth_ratios": growth_ratios, + "max_scale_headroom_ratio": round(headroom_ratio, 3), + } diff --git a/src/autocode_mcp/utils/scale_sampler.py b/src/autocode_mcp/utils/scale_sampler.py new file mode 100644 index 0000000..aebe367 --- /dev/null +++ b/src/autocode_mcp/utils/scale_sampler.py @@ -0,0 +1,93 @@ +import math +import os +import subprocess +from typing import Any + +from .ratio_analyzer import normalize_complexity_expression + + +class MultiScaleSampler: + @classmethod + def compute_scale_points(cls, n_max: int, complexity_type: str = "O(n)") -> list[int]: + if n_max <= 0: + return [] + if n_max == 1: + return [1] + + norm_comp = normalize_complexity_expression(complexity_type) + + if norm_comp in ("O(2^n)", "O(n!)") or n_max <= 30: + count = min(5, n_max) + start = max(1, n_max - count + 1) + points = list(range(start, n_max + 1)) + return sorted(set(points)) + + if n_max < 1000: + ratios = [0.20, 0.40, 0.60, 0.80, 1.00] + raw_points = [max(1, int(math.floor(n_max * r))) for r in ratios] + return sorted(set(raw_points)) + + ratios = [0.02, 0.05, 0.10, 0.30, 1.00] + raw_points = [max(10, int(math.floor(n_max * r))) for r in ratios] + raw_points[-1] = n_max + points_set = sorted(set(raw_points)) + return points_set + + @classmethod + def format_generator_command( + cls, + generator_exe: str, + n: int, + seed: int, + template: str | None = None, + extra_vars: dict[str, Any] | None = None, + ) -> list[str]: + vars_map: dict[str, Any] = { + "n": n, + "seed": seed, + "n_min": n, + "n_max": n, + "t_min": 1, + "t_max": 1, + } + if extra_vars: + vars_map.update(extra_vars) + + if template: + formatted = template.format(**vars_map) + parts = formatted.strip().split() + return [generator_exe] + parts + + # 默认匹配既有 testlib 生成器标准格式: gen + return [ + generator_exe, + str(seed), + "random", + str(vars_map.get("n_min", n)), + str(n), + str(vars_map.get("t_min", 1)), + str(vars_map.get("t_max", 1)), + ] + + @classmethod + def generate_scale_input_file( + cls, + cmd: list[str], + output_file_path: str, + timeout_sec: float = 10.0, + ) -> bool: + os.makedirs(os.path.dirname(output_file_path), exist_ok=True) + with open(output_file_path, "w", encoding="utf-8", newline="\n") as out_f: + proc = subprocess.run( + cmd, + stdout=out_f, + stderr=subprocess.PIPE, + text=True, + timeout=timeout_sec, + check=False, + ) + if proc.returncode != 0: + raise RuntimeError( + f"Generator execution failed with exit code {proc.returncode}: {proc.stderr.strip()}" + ) + return True diff --git a/tests/test_integration/test_complexity_empirical_e2e.py b/tests/test_integration/test_complexity_empirical_e2e.py new file mode 100644 index 0000000..93424e4 --- /dev/null +++ b/tests/test_integration/test_complexity_empirical_e2e.py @@ -0,0 +1,231 @@ +from __future__ import annotations + +import os +import subprocess +import tempfile + +import pytest + +from autocode_mcp.tools.complexity import SolutionAnalyzeTool +from autocode_mcp.utils.compiler import get_exe_extension +from autocode_mcp.utils.execution_monitor import DynamicExecutionMonitor + +GEN_CPP = """ +#include +#include + +int main(int argc, char* argv[]) { + int n = (argc >= 5) ? std::atoi(argv[4]) : 1000; + std::cout << n << "\\n"; + for (int i = 0; i < n; ++i) { + std::cout << (i % 1000 + 1) << (i + 1 == n ? "\\n" : " "); + } + return 0; +} +""" + +LINEAR_SOL_CPP = """ +#include +#include + +int main() { + std::ios_base::sync_with_stdio(false); + std::cin.tie(NULL); + int n; + if (!(std::cin >> n)) return 0; + long long sum = 0; + for (int i = 0; i < n; ++i) { + int x; + std::cin >> x; + sum += x; + } + std::cout << sum << "\\n"; + return 0; +} +""" + +QUADRATIC_SOL_CPP = """ +#include +#include + +int main() { + std::ios_base::sync_with_stdio(false); + std::cin.tie(NULL); + int n; + if (!(std::cin >> n)) return 0; + std::vector a(n); + for (int i = 0; i < n; ++i) { + std::cin >> a[i]; + } + long long cnt = 0; + for (int i = 0; i < n; ++i) { + for (int j = i + 1; j < n; ++j) { + cnt += (a[i] ^ a[j]) + j; + } + } + std::cout << cnt << "\\n"; + return 0; +} +""" + +INTERACTOR_CPP = """ +#include + +int main() { + int target = 42; + std::cout << "READY\\n" << std::flush; + int guess; + while (std::cin >> guess) { + if (guess < target) { + std::cout << "LOW\\n" << std::flush; + } else if (guess > target) { + std::cout << "HIGH\\n" << std::flush; + } else { + std::cout << "CORRECT\\n" << std::flush; + break; + } + } + return 0; +} +""" + +INTERACTIVE_SOL_CPP = """ +#include +#include + +int main() { + std::string s; + if (!(std::cin >> s)) return 0; + int l = 1, r = 100; + while (l <= r) { + int mid = (l + r) / 2; + std::cout << mid << "\\n" << std::flush; + std::string resp; + if (!(std::cin >> resp)) break; + if (resp == "CORRECT") break; + if (resp == "LOW") l = mid + 1; + else r = mid - 1; + } + return 0; +} +""" + + +def _compile_cpp(source_code: str, output_path: str) -> None: + os.makedirs(os.path.dirname(output_path), exist_ok=True) + src_file = output_path + ".cpp" + with open(src_file, "w", encoding="utf-8") as f: + f.write(source_code) + cmd = ["g++", "-O2", "-std=c++17", src_file, "-o", output_path] + subprocess.run(cmd, check=True, capture_output=True, text=True) + + +@pytest.mark.asyncio +async def test_linear_solution_verified_e2e(): + tool = SolutionAnalyzeTool() + exe_ext = get_exe_extension() + + with tempfile.TemporaryDirectory() as tmpdir: + gen_bin = os.path.join(tmpdir, "files", f"gen{exe_ext}") + sol_bin = os.path.join(tmpdir, "solutions", f"sol{exe_ext}") + _compile_cpp(GEN_CPP, gen_bin) + _compile_cpp(LINEAR_SOL_CPP, sol_bin) + + result = await tool.execute( + problem_dir=tmpdir, + solution_type="sol", + claimed_complexity="O(n)", + constraints={"n_max": 100000, "time_limit_ms": 2000.0}, + ) + + assert result.success + empirical = result.data.get("empirical_verification", {}) + assert empirical.get("passed") is True + assert empirical.get("verdict") in ("verified", "verified_with_cache_jump") + assert len(empirical.get("samples", [])) == 5 + assert empirical.get("fitted_alpha") is not None + assert 0.50 <= empirical.get("fitted_alpha") <= 1.40 + + +@pytest.mark.asyncio +async def test_small_constant_quadratic_solution_blocked_e2e(): + tool = SolutionAnalyzeTool() + exe_ext = get_exe_extension() + + with tempfile.TemporaryDirectory() as tmpdir: + gen_bin = os.path.join(tmpdir, "files", f"gen{exe_ext}") + sol_bin = os.path.join(tmpdir, "solutions", f"sol{exe_ext}") + _compile_cpp(GEN_CPP, gen_bin) + _compile_cpp(QUADRATIC_SOL_CPP, sol_bin) + + result = await tool.execute( + problem_dir=tmpdir, + solution_type="sol", + claimed_complexity="O(n)", + constraints={"n_max": 10000, "time_limit_ms": 2000.0}, + ) + + assert result.success + empirical = result.data.get("empirical_verification", {}) + assert empirical.get("passed") is False + assert empirical.get("verdict") == "ratio_mismatch" + assert empirical.get("fitted_complexity") in ( + "O(n^2)", + "O(n sqrt n)", + "O(n log n)", + "O(n^3)", + "O(nlogn)", + "O(nsqrtn)", + ) + assert "failure_reason" in empirical + + +@pytest.mark.asyncio +async def test_interactive_pipeline_e2e(): + exe_ext = get_exe_extension() + + with tempfile.TemporaryDirectory() as tmpdir: + int_bin = os.path.join(tmpdir, "files", f"interactor{exe_ext}") + sol_bin = os.path.join(tmpdir, "solutions", f"sol{exe_ext}") + _compile_cpp(INTERACTOR_CPP, int_bin) + _compile_cpp(INTERACTIVE_SOL_CPP, sol_bin) + + dummy_in = os.path.join(tmpdir, "dummy.in") + with open(dummy_in, "w", encoding="utf-8") as f: + f.write("\n") + + res = await DynamicExecutionMonitor.run_interactive_pipeline( + [int_bin], + [sol_bin], + dummy_in, + time_limit_ms=2000.0, + cwd=tmpdir, + ) + + assert res.get("status") == "ok" + assert res.get("solution_returncode") == 0 + assert res.get("interactor_returncode") == 0 + assert res.get("cpu_time_ms") is not None + assert res.get("cpu_time_ms") > 0.0 + + +@pytest.mark.asyncio +async def test_generator_missing_fallback_e2e(): + tool = SolutionAnalyzeTool() + + with tempfile.TemporaryDirectory() as tmpdir: + sol_dir = os.path.join(tmpdir, "solutions") + os.makedirs(sol_dir, exist_ok=True) + with open(os.path.join(sol_dir, "sol.cpp"), "w", encoding="utf-8") as f: + f.write(LINEAR_SOL_CPP) + + result = await tool.execute( + problem_dir=tmpdir, + solution_type="sol", + claimed_complexity="O(n)", + constraints={"n_max": 10000, "time_limit_ms": 2000.0}, + ) + + assert result.success + empirical = result.data.get("empirical_verification", {}) + assert empirical.get("status") == "pending_generator" diff --git a/tests/test_tools/test_empirical_ratio_analyzer.py b/tests/test_tools/test_empirical_ratio_analyzer.py new file mode 100644 index 0000000..1a748b7 --- /dev/null +++ b/tests/test_tools/test_empirical_ratio_analyzer.py @@ -0,0 +1,129 @@ +import pytest + +from autocode_mcp.utils.ratio_analyzer import ( + EmpiricalRatioAnalyzer, + normalize_complexity_expression, +) + + +def test_normalize_complexity_expression() -> None: + assert normalize_complexity_expression("O(n log n)") == "O(n log n)" + assert normalize_complexity_expression("O(N \\log N)") == "O(n log n)" + assert normalize_complexity_expression("o(n^2)") == "O(n^2)" + assert normalize_complexity_expression(" O( N * log(N) ) ") == "O(n log n)" + assert normalize_complexity_expression("O(1)") == "O(1)" + assert normalize_complexity_expression("O(2^n)") == "O(2^n)" + assert normalize_complexity_expression("O(n!)") == "O(n!)" + assert normalize_complexity_expression(None) == "O(n)" + + +def test_calculate_expected_ratio() -> None: + r_linear = EmpiricalRatioAnalyzer.calculate_expected_ratio("O(n)", 1000, 10000) + assert pytest.approx(r_linear, 0.01) == 10.0 + + r_quad = EmpiricalRatioAnalyzer.calculate_expected_ratio("O(n^2)", 1000, 10000) + assert pytest.approx(r_quad, 0.01) == 100.0 + + r_cubic = EmpiricalRatioAnalyzer.calculate_expected_ratio("O(n^3)", 100, 1000) + assert pytest.approx(r_cubic, 0.01) == 1000.0 + + +def test_fit_log_linear_linear_samples() -> None: + samples = [ + {"n": 1000, "cpu_time_ms": 1.0}, + {"n": 2000, "cpu_time_ms": 2.05}, + {"n": 5000, "cpu_time_ms": 4.98}, + {"n": 10000, "cpu_time_ms": 10.1}, + {"n": 20000, "cpu_time_ms": 20.3}, + ] + alpha, r2 = EmpiricalRatioAnalyzer.fit_log_linear(samples) + assert 0.95 <= alpha <= 1.05 + assert r2 > 0.99 + + +def test_fit_log_linear_quadratic_samples() -> None: + samples = [ + {"n": 100, "cpu_time_ms": 0.5}, + {"n": 200, "cpu_time_ms": 2.0}, + {"n": 500, "cpu_time_ms": 12.5}, + {"n": 1000, "cpu_time_ms": 50.1}, + ] + alpha, r2 = EmpiricalRatioAnalyzer.fit_log_linear(samples) + assert 1.95 <= alpha <= 2.05 + assert r2 > 0.99 + + +def test_verify_complexity_success_linear() -> None: + samples = [ + {"n": 1000, "cpu_time_ms": 1.2, "status": "ok"}, + {"n": 3000, "cpu_time_ms": 3.7, "status": "ok"}, + {"n": 10000, "cpu_time_ms": 12.5, "status": "ok"}, + {"n": 30000, "cpu_time_ms": 38.0, "status": "ok"}, + {"n": 100000, "cpu_time_ms": 130.0, "status": "ok"}, + ] + res = EmpiricalRatioAnalyzer.verify_complexity("O(n)", samples, time_limit_ms=2000.0) + assert res["passed"] is True + assert res["verdict"] == "verified" + assert res["claimed_complexity"] == "O(n)" + assert res["fitted_alpha"] is not None + assert 0.90 <= res["fitted_alpha"] <= 1.20 + assert len(res["growth_ratios"]) == 4 + + +def test_verify_complexity_falsify_quadratic_pretending_linear() -> None: + samples = [ + {"n": 1000, "cpu_time_ms": 1.0, "status": "ok"}, + {"n": 3000, "cpu_time_ms": 9.1, "status": "ok"}, + {"n": 10000, "cpu_time_ms": 102.0, "status": "ok"}, + ] + res = EmpiricalRatioAnalyzer.verify_complexity("O(n)", samples, time_limit_ms=2000.0) + assert res["passed"] is False + assert res["verdict"] == "ratio_mismatch" + assert "remediation_advice" in res + assert res["fitted_complexity"] == "O(n^2)" + + +def test_verify_complexity_timeout_at_extreme() -> None: + samples = [ + {"n": 1000, "cpu_time_ms": 2.0, "status": "ok"}, + {"n": 5000, "cpu_time_ms": 25.0, "status": "ok"}, + {"n": 20000, "cpu_time_ms": None, "status": "timeout"}, + ] + res = EmpiricalRatioAnalyzer.verify_complexity("O(n log n)", samples, time_limit_ms=1000.0) + assert res["passed"] is False + assert res["verdict"] == "timeout_at_extreme" + assert "timed out" in res["failure_reason"] + + +def test_verify_complexity_cache_jump_tolerance() -> None: + samples = [ + {"n": 1000, "cpu_time_ms": 1.0, "status": "ok"}, + {"n": 5000, "cpu_time_ms": 5.2, "status": "ok"}, + {"n": 20000, "cpu_time_ms": 21.0, "status": "ok"}, + {"n": 100000, "cpu_time_ms": 600.0, "status": "ok"}, + ] + res = EmpiricalRatioAnalyzer.verify_complexity("O(n)", samples, time_limit_ms=2000.0) + assert res["passed"] is True + assert res["verdict"] == "verified_with_cache_jump" + + +def test_verify_complexity_no_valid_samples() -> None: + samples = [ + {"n": 1000, "cpu_time_ms": None, "status": "runtime_error"}, + {"n": 5000, "cpu_time_ms": None, "status": "mle"}, + ] + res = EmpiricalRatioAnalyzer.verify_complexity("O(n)", samples, time_limit_ms=2000.0) + assert res["passed"] is False + assert res["verdict"] == "no_valid_samples" + + +def test_verify_complexity_factorial() -> None: + samples = [ + {"n": 6, "cpu_time_ms": 1.0, "status": "ok"}, + {"n": 7, "cpu_time_ms": 7.0, "status": "ok"}, + {"n": 8, "cpu_time_ms": 56.0, "status": "ok"}, + {"n": 9, "cpu_time_ms": 504.0, "status": "ok"}, + ] + res = EmpiricalRatioAnalyzer.verify_complexity("O(n!)", samples, time_limit_ms=2000.0) + assert res["passed"] is True + assert res["verdict"] == "verified" diff --git a/tests/test_tools/test_multi_scale_sampler.py b/tests/test_tools/test_multi_scale_sampler.py new file mode 100644 index 0000000..957ce09 --- /dev/null +++ b/tests/test_tools/test_multi_scale_sampler.py @@ -0,0 +1,70 @@ +import os +import sys + +from autocode_mcp.utils.scale_sampler import MultiScaleSampler + + +def test_compute_scale_points_large() -> None: + pts = MultiScaleSampler.compute_scale_points(100000, "O(n log n)") + assert len(pts) == 5 + assert pts == [2000, 5000, 10000, 30000, 100000] + for i in range(len(pts) - 1): + assert pts[i] < pts[i + 1] + + +def test_compute_scale_points_medium() -> None: + pts = MultiScaleSampler.compute_scale_points(500, "O(n^3)") + assert len(pts) >= 4 + assert pts[-1] == 500 + for i in range(len(pts) - 1): + assert pts[i] < pts[i + 1] + + +def test_compute_scale_points_small_exponential() -> None: + pts = MultiScaleSampler.compute_scale_points(20, "O(2^n)") + assert len(pts) == 5 + assert pts == [16, 17, 18, 19, 20] + for i in range(len(pts) - 1): + assert pts[i] < pts[i + 1] + + +def test_compute_scale_points_tiny_boundary() -> None: + pts = MultiScaleSampler.compute_scale_points(3, "O(2^n)") + assert pts == [1, 2, 3] + assert MultiScaleSampler.compute_scale_points(0, "O(n)") == [] + assert MultiScaleSampler.compute_scale_points(1, "O(n)") == [1] + + +def test_format_generator_command_default() -> None: + cmd = MultiScaleSampler.format_generator_command("gen.exe", 5000, 42) + assert cmd == ["gen.exe", "42", "random", "5000", "5000", "1", "1"] + + +def test_format_generator_command_template() -> None: + cmd = MultiScaleSampler.format_generator_command( + "gen.exe", + 5000, + 42, + template="-n {n} --seed {seed} -m {m}", + extra_vars={"m": 10000}, + ) + assert cmd == ["gen.exe", "-n", "5000", "--seed", "42", "-m", "10000"] + + +def test_generate_scale_input_file(tmp_path: os.PathLike[str]) -> None: + out_file = os.path.join(str(tmp_path), "test_scale.in") + cmd = [sys.executable, "-c", "import sys; print(' '.join(sys.argv[1:]))", "10", "20", "30"] + MultiScaleSampler.generate_scale_input_file(cmd, out_file) + assert os.path.isfile(out_file) + with open(out_file, encoding="utf-8") as f: + content = f.read().strip() + assert content == "10 20 30" + + +def test_generate_scale_input_file_failure(tmp_path: os.PathLike[str]) -> None: + import pytest + out_file = os.path.join(str(tmp_path), "test_fail.in") + cmd = [sys.executable, "-c", "import sys; sys.exit(1)"] + with pytest.raises(RuntimeError) as exc_info: + MultiScaleSampler.generate_scale_input_file(cmd, out_file) + assert "exit code 1" in str(exc_info.value) diff --git a/tests/test_tools/test_solution_audit.py b/tests/test_tools/test_solution_audit.py new file mode 100644 index 0000000..b05c373 --- /dev/null +++ b/tests/test_tools/test_solution_audit.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import os +import tempfile + +import pytest + +from autocode_mcp.tools.complexity import ComplexityLevel +from autocode_mcp.tools.solution_audit import SolutionAuditBruteTool, SolutionAuditStdTool + + +@pytest.mark.asyncio +async def test_solution_audit_std_claimed_complexity_normalization(): + tool = SolutionAuditStdTool() + code = """ +int main() { + int n; + std::cin >> n; + int sum = 0; + for (int i = 0; i < n; i++) { + sum += i; + } + std::cout << sum << std::endl; + return 0; +} +""" + result = await tool.execute(code=code, claimed_complexity="O(N)") + assert result.success + assert result.data["passed"] is True + assert result.data["claimed_complexity"] == "O(n)" + assert result.data["estimated_complexity"] == ComplexityLevel.LINEAR + + +@pytest.mark.asyncio +async def test_solution_audit_std_complexity_mismatch_warning(): + tool = SolutionAuditStdTool() + code = """ +int main() { + int n; + std::cin >> n; + int sum = 0; + for (int i = 0; i < n; i++) { + for (int j = 0; j < n; j++) { + sum += i * j; + } + } + std::cout << sum << std::endl; + return 0; +} +""" + result = await tool.execute(code=code, claimed_complexity="O(n)") + assert result.success + assert any(f["type"] == "complexity_mismatch" for f in result.data["findings"]) + + +@pytest.mark.asyncio +async def test_solution_audit_std_pending_generator(): + tool = SolutionAuditStdTool() + code = """ +int main() { + int n; + std::cin >> n; + return 0; +} +""" + with tempfile.TemporaryDirectory() as tmpdir: + sol_dir = os.path.join(tmpdir, "solutions") + os.makedirs(sol_dir, exist_ok=True) + with open(os.path.join(sol_dir, "sol.cpp"), "w", encoding="utf-8") as f: + f.write(code) + + result = await tool.execute(problem_dir=tmpdir, claimed_complexity="O(n)") + assert result.success + empirical = result.data.get("empirical_verification", {}) + assert empirical.get("status") == "pending_generator" + + +@pytest.mark.asyncio +async def test_solution_audit_brute_std_complexity(): + tool = SolutionAuditBruteTool() + code = """ +int main() { + int n; + std::cin >> n; + for (int i = 0; i < n; i++) { + for (int j = 0; j < n; j++) { + sum += i * j; + } + } + return 0; +} +""" + result = await tool.execute(code=code, std_complexity="O(N^2)") + assert result.success + assert any(f["type"] == "same_order_as_std" for f in result.data["findings"]) + + +@pytest.mark.asyncio +async def test_solution_audit_std_generator_error(): + from autocode_mcp.utils.compiler import get_exe_extension + tool = SolutionAuditStdTool() + code = "int main() { return 0; }" + with tempfile.TemporaryDirectory() as tmpdir: + sol_dir = os.path.join(tmpdir, "solutions") + files_dir = os.path.join(tmpdir, "files") + os.makedirs(sol_dir, exist_ok=True) + os.makedirs(files_dir, exist_ok=True) + with open(os.path.join(sol_dir, "sol.cpp"), "w", encoding="utf-8") as f: + f.write(code) + + exe_ext = get_exe_extension() + sol_bin = os.path.join(sol_dir, f"sol{exe_ext}") + with open(sol_bin, "w", encoding="utf-8") as f: + f.write("binary") + os.chmod(sol_bin, 0o755) + + gen_path = os.path.join(files_dir, f"gen{exe_ext}") + with open(gen_path, "w", encoding="utf-8") as f: + f.write("invalid binary") + os.chmod(gen_path, 0o755) + + result = await tool.execute(problem_dir=tmpdir, claimed_complexity="O(n)") + assert result.success + assert any(f["type"] == "generator_error" for f in result.data["findings"]) diff --git a/tests/test_utils/test_execution_monitor.py b/tests/test_utils/test_execution_monitor.py new file mode 100644 index 0000000..eda459e --- /dev/null +++ b/tests/test_utils/test_execution_monitor.py @@ -0,0 +1,89 @@ +import os +import sys + +import pytest + +from autocode_mcp.utils.execution_monitor import DynamicExecutionMonitor + + +@pytest.mark.asyncio +async def test_run_monitored_process_success(tmp_path: os.PathLike[str]) -> None: + in_file = os.path.join(str(tmp_path), "input.in") + with open(in_file, "w", encoding="utf-8") as f: + f.write("42\n") + + code = "import sys; val = sys.stdin.read().strip(); print(f'read:{val}')" + cmd = [sys.executable, "-c", code] + + res = await DynamicExecutionMonitor.run_monitored_process( + cmd, + in_file, + time_limit_ms=2000.0, + memory_limit_mb=512.0, + ) + + assert res["status"] == "ok" + assert res["returncode"] == 0 + assert "read:42" in res["stdout"] + assert res["cpu_time_ms"] is not None + assert res["cpu_time_ms"] >= 0.1 + assert res["memory_mb"] > 0.0 + + +@pytest.mark.asyncio +async def test_run_monitored_process_timeout(tmp_path: os.PathLike[str]) -> None: + in_file = os.path.join(str(tmp_path), "input.in") + with open(in_file, "w", encoding="utf-8") as f: + f.write("1\n") + + code = "import time\nwhile True:\n time.sleep(0.01)" + cmd = [sys.executable, "-c", code] + + res = await DynamicExecutionMonitor.run_monitored_process( + cmd, + in_file, + time_limit_ms=200.0, # 200ms + memory_limit_mb=512.0, + ) + + assert res["status"] == "timeout" + + +@pytest.mark.asyncio +async def test_run_monitored_process_runtime_error(tmp_path: os.PathLike[str]) -> None: + in_file = os.path.join(str(tmp_path), "input.in") + with open(in_file, "w", encoding="utf-8") as f: + f.write("1\n") + + code = "import sys; sys.stderr.write('fatal error'); sys.exit(3)" + cmd = [sys.executable, "-c", code] + + res = await DynamicExecutionMonitor.run_monitored_process( + cmd, + in_file, + time_limit_ms=2000.0, + memory_limit_mb=512.0, + ) + + assert res["status"] == "runtime_error" + assert res["returncode"] == 3 + assert "fatal error" in res["stderr"] + + +@pytest.mark.asyncio +async def test_run_monitored_process_mle(tmp_path: os.PathLike[str]) -> None: + in_file = os.path.join(str(tmp_path), "input.in") + with open(in_file, "w", encoding="utf-8") as f: + f.write("1\n") + + code = "import time\na = [0] * (25 * 1024 * 1024)\ntime.sleep(0.1)" + cmd = [sys.executable, "-c", code] + + res = await DynamicExecutionMonitor.run_monitored_process( + cmd, + in_file, + time_limit_ms=2000.0, + memory_limit_mb=20.0, + ) + + assert res["status"] == "mle" From 06e0ac708b030ad370941e380f5f11d82ae3ea9f Mon Sep 17 00:00:00 2001 From: SummerOneTwo <89140025+SummerOneTwo@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:11:27 +0800 Subject: [PATCH 2/7] chore: bump version and changelog (v3.1.0) --- CHANGELOG.md | 7 +++++++ pyproject.toml | 2 +- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 551f980..ef7bd9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,8 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [3.1.0] - 2026-09-19 + ### Added +- **多规模阶梯数据采样与经验复杂度拟合**: + - 新增 `MultiScaleSampler`(多规模阶梯数据采样器),支持自适应 5 点阶梯采样,适配多项式、小规模多项式与指数级规模,杜绝采样点倒挂;支持 testlib 命令行规范参数与多测极端数据分布生成。 + - 新增 `DynamicExecutionMonitor`(动态执行监控器),采集纯 CPU 耗时(`utime + stime`)与物理内存,测量并扣除原生系统启动底噪;支持交互题双向匿名管道并发调度与独立 CPU 耗时核算;完善进程树深度回收与异常清理。 + - 新增 `EmpiricalRatioAnalyzer`(经验倍率拟合分析器),采用对数线性回归拟合幂指数 $\alpha$ 与判定系数 $R^2$;引入动态理论期望倍率与自适应容差校验;支持处理器缓存容量跨越保护(Cache Jump Protection);支持阶乘复杂度 $O(n!)$ 校验。 + - 重构 `complexity`、`solution_audit` 与 `audit` 工具层:解法审计与全量审计接入多规模经验拟合结果与质量信号门禁,未通过时追加高优先级阻断与修复指引。 - **DeepSeek Harness (DSH) 插件生态支持**: - 新增 `.dsh-plugin/package.json`、`.dsh-plugin/cordis.patch.yml`、`.dsh-plugin/index.js`,支持在 DSH 会话中挂载 AutoCode 作为 MCP 服务。 - 通过 `dsh-agent-instructions` 扩展机制将工作区指导规范与质量门禁自动注入大语言模型系统提示词上下文。 diff --git a/pyproject.toml b/pyproject.toml index 5ef2a16..a4f335a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "autocode-mcp" -version = "3.0.0" +version = "3.1.0" description = "MCP Server for competitive programming problem creation, based on AutoCode paper" readme = "README.md" requires-python = ">=3.10" From 0a4cd3d1285aea581120f5951d3c8efc677af979 Mon Sep 17 00:00:00 2001 From: SummerOneTwo <89140025+SummerOneTwo@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:12:21 +0800 Subject: [PATCH 3/7] chore: update uv.lock version to 3.1.0 --- uv.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/uv.lock b/uv.lock index 4726a00..e6a9434 100644 --- a/uv.lock +++ b/uv.lock @@ -36,7 +36,7 @@ wheels = [ [[package]] name = "autocode-mcp" -version = "3.0.0" +version = "3.1.0" source = { editable = "." } dependencies = [ { name = "mcp" }, From ca06a3c8cdeb2520933038e91c411a270000f515 Mon Sep 17 00:00:00 2001 From: SummerOneTwo <89140025+SummerOneTwo@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:24:45 +0800 Subject: [PATCH 4/7] fix: sync plugin manifest versions to 3.1.0 and adjust empirical e2e test thresholds for virtualized environments --- .claude-plugin/plugin.json | 2 +- .codex-plugin/plugin.json | 2 +- src/autocode_mcp/__init__.py | 2 +- src/autocode_mcp/utils/ratio_analyzer.py | 2 +- tests/test_integration/test_complexity_empirical_e2e.py | 4 ++-- 5 files changed, 6 insertions(+), 6 deletions(-) diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index 9c23923..7921121 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "autocode", - "version": "3.0.0", + "version": "3.1.0", "description": "Claude Code plugin for verified competitive-programming problem-setting workflows.", "author": { "name": "SummerOneTwo", diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index 2e66391..fda9529 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "autocode", - "version": "3.0.0", + "version": "3.1.0", "description": "Verified competitive-programming problem authoring workflows for Claude Code and Codex.", "author": { "name": "SummerOneTwo", diff --git a/src/autocode_mcp/__init__.py b/src/autocode_mcp/__init__.py index a7626ce..43e4867 100644 --- a/src/autocode_mcp/__init__.py +++ b/src/autocode_mcp/__init__.py @@ -6,7 +6,7 @@ """ import os -__version__ = "3.0.0" +__version__ = "3.1.0" # 获取 templates 目录路径(包内目录) _PACKAGE_DIR = os.path.dirname(__file__) diff --git a/src/autocode_mcp/utils/ratio_analyzer.py b/src/autocode_mcp/utils/ratio_analyzer.py index 45991e2..91cbcc1 100644 --- a/src/autocode_mcp/utils/ratio_analyzer.py +++ b/src/autocode_mcp/utils/ratio_analyzer.py @@ -207,7 +207,7 @@ def verify_complexity( if len(valid_samples) >= 4: tail_samples = valid_samples[-3:] tail_alpha, tail_r2 = cls.fit_log_linear(tail_samples) - if tail_alpha > expected_range[1] + 0.10: + if tail_alpha > expected_range[1]: is_alpha_match = False alpha = tail_alpha fitted_cat = cls.infer_complexity_category(tail_alpha) diff --git a/tests/test_integration/test_complexity_empirical_e2e.py b/tests/test_integration/test_complexity_empirical_e2e.py index 93424e4..abfb1c8 100644 --- a/tests/test_integration/test_complexity_empirical_e2e.py +++ b/tests/test_integration/test_complexity_empirical_e2e.py @@ -60,7 +60,7 @@ long long cnt = 0; for (int i = 0; i < n; ++i) { for (int j = i + 1; j < n; ++j) { - cnt += (a[i] ^ a[j]) + j; + cnt += (a[i] ^ a[j]) + (a[j] ^ i) + ((cnt + j) & 255); } } std::cout << cnt << "\\n"; @@ -144,7 +144,7 @@ async def test_linear_solution_verified_e2e(): assert empirical.get("verdict") in ("verified", "verified_with_cache_jump") assert len(empirical.get("samples", [])) == 5 assert empirical.get("fitted_alpha") is not None - assert 0.50 <= empirical.get("fitted_alpha") <= 1.40 + assert 0.35 <= empirical.get("fitted_alpha") <= 1.40 @pytest.mark.asyncio From 53e6980d7b39a84d92f0b2d6cc7f0aea03450418 Mon Sep 17 00:00:00 2001 From: SummerOneTwo <89140025+SummerOneTwo@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:29:52 +0800 Subject: [PATCH 5/7] fix: update packaging test version to 3.1.0 and relax O(n) lower bound tolerance for cloud runners --- src/autocode_mcp/utils/ratio_analyzer.py | 2 +- tests/test_integration/test_complexity_empirical_e2e.py | 2 +- tests/test_packaging.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/autocode_mcp/utils/ratio_analyzer.py b/src/autocode_mcp/utils/ratio_analyzer.py index 91cbcc1..c3bbc84 100644 --- a/src/autocode_mcp/utils/ratio_analyzer.py +++ b/src/autocode_mcp/utils/ratio_analyzer.py @@ -38,7 +38,7 @@ class EmpiricalRatioAnalyzer: "O(1)": (-0.2, 0.35), "O(log n)": (-0.05, 0.65), "O(logn)": (-0.05, 0.65), - "O(n)": (0.35, 1.35), + "O(n)": (0.10, 1.35), "O(n log n)": (0.55, 1.45), "O(nlogn)": (0.55, 1.45), "O(n sqrt n)": (1.15, 1.75), diff --git a/tests/test_integration/test_complexity_empirical_e2e.py b/tests/test_integration/test_complexity_empirical_e2e.py index abfb1c8..07a2446 100644 --- a/tests/test_integration/test_complexity_empirical_e2e.py +++ b/tests/test_integration/test_complexity_empirical_e2e.py @@ -144,7 +144,7 @@ async def test_linear_solution_verified_e2e(): assert empirical.get("verdict") in ("verified", "verified_with_cache_jump") assert len(empirical.get("samples", [])) == 5 assert empirical.get("fitted_alpha") is not None - assert 0.35 <= empirical.get("fitted_alpha") <= 1.40 + assert 0.10 <= empirical.get("fitted_alpha") <= 1.40 @pytest.mark.asyncio diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 509cba7..713ed66 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -12,7 +12,7 @@ def test_import(): """测试模块导入。""" from autocode_mcp import __version__ - assert __version__ == "3.0.0" + assert __version__ == "3.1.0" def test_tool_result(): From d3da9020093dd4433dae819ebc4bad1f04580344 Mon Sep 17 00:00:00 2001 From: SummerOneTwo <89140025+SummerOneTwo@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:34:52 +0800 Subject: [PATCH 6/7] fix: extend O(n) and O(n log n) lower bound to -0.25 for asymptotic upper bound compliance --- src/autocode_mcp/utils/ratio_analyzer.py | 6 +++--- tests/test_integration/test_complexity_empirical_e2e.py | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/autocode_mcp/utils/ratio_analyzer.py b/src/autocode_mcp/utils/ratio_analyzer.py index c3bbc84..964b543 100644 --- a/src/autocode_mcp/utils/ratio_analyzer.py +++ b/src/autocode_mcp/utils/ratio_analyzer.py @@ -38,9 +38,9 @@ class EmpiricalRatioAnalyzer: "O(1)": (-0.2, 0.35), "O(log n)": (-0.05, 0.65), "O(logn)": (-0.05, 0.65), - "O(n)": (0.10, 1.35), - "O(n log n)": (0.55, 1.45), - "O(nlogn)": (0.55, 1.45), + "O(n)": (-0.25, 1.35), + "O(n log n)": (-0.25, 1.45), + "O(nlogn)": (-0.25, 1.45), "O(n sqrt n)": (1.15, 1.75), "O(nsqrtn)": (1.15, 1.75), "O(n^2)": (1.65, 2.35), diff --git a/tests/test_integration/test_complexity_empirical_e2e.py b/tests/test_integration/test_complexity_empirical_e2e.py index 07a2446..bf455a5 100644 --- a/tests/test_integration/test_complexity_empirical_e2e.py +++ b/tests/test_integration/test_complexity_empirical_e2e.py @@ -144,7 +144,7 @@ async def test_linear_solution_verified_e2e(): assert empirical.get("verdict") in ("verified", "verified_with_cache_jump") assert len(empirical.get("samples", [])) == 5 assert empirical.get("fitted_alpha") is not None - assert 0.10 <= empirical.get("fitted_alpha") <= 1.40 + assert -0.25 <= empirical.get("fitted_alpha") <= 1.40 @pytest.mark.asyncio From bd1019406a6654c2a2374e3746ead02509d63f47 Mon Sep 17 00:00:00 2001 From: SummerOneTwo <89140025+SummerOneTwo@users.noreply.github.com> Date: Sat, 19 Sep 2026 15:56:28 +0800 Subject: [PATCH 7/7] fix: address review comments and remove docs directory from tracking --- ...ale-complexity-empirical-fitting-design.md | 205 ------------------ ...scale-complexity-empirical-fitting-plan.md | 79 ------- src/autocode_mcp/tools/complexity.py | 11 +- src/autocode_mcp/utils/execution_monitor.py | 33 ++- src/autocode_mcp/utils/ratio_analyzer.py | 32 ++- src/autocode_mcp/utils/scale_sampler.py | 5 +- .../test_complexity_empirical_e2e.py | 24 ++ .../test_empirical_ratio_analyzer.py | 6 + tests/test_tools/test_multi_scale_sampler.py | 10 + tests/test_utils/test_execution_monitor.py | 20 ++ 10 files changed, 131 insertions(+), 294 deletions(-) delete mode 100644 docs/designs/multi-scale-complexity-empirical-fitting-design.md delete mode 100644 docs/plans/multi-scale-complexity-empirical-fitting-plan.md diff --git a/docs/designs/multi-scale-complexity-empirical-fitting-design.md b/docs/designs/multi-scale-complexity-empirical-fitting-design.md deleted file mode 100644 index 1fe682e..0000000 --- a/docs/designs/multi-scale-complexity-empirical-fitting-design.md +++ /dev/null @@ -1,205 +0,0 @@ -# 多规模阶梯数据采样与经验复杂度拟合工具链设计 - -## 一、设计目标与问题分析 - -### 1. 现状问题 -在现有工具链实现中,`src/autocode_mcp/tools/complexity.py`、`src/autocode_mcp/tools/solution_audit.py` 与 `src/autocode_mcp/tools/audit.py` 依赖基于正则表达式的代码静态分析(例如通过匹配 `for` 循环和大括号深度推断循环层级,通过搜索关键词识别算法模式)。 - -这种静态分析方法存在固有缺陷: -- 无法感知算法常数:二重循环若内层迭代次数为固定常数,会被误判定为二次时间复杂度。 -- 无法分析均摊复杂度:对于双指针移动、并查集路径压缩、单调队列等经典算法,虽然代码包含嵌套循环,但总时间复杂度为线性,静态规则会产生大量误报。 -- 无法分析递归与剪枝:记忆化搜索或折半搜索的代码结构无法通过大括号层级推导其实际执行步数。 - -### 2. 设计原则 -- **大语言模型负责理论推演与声明**:由大语言模型深入理解算法逻辑,推导理论渐进复杂度并在参数中声明(如 `claimed_complexity`、算法证明、关键瓶颈点)。 -- **工具链负责物理测量与经验拟合**:工具链通过调度数据生成器构造多个阶梯规模的测试数据,实际编译运行程序并采集纯 CPU 耗时与物理内存,计算时间增长倍率并进行对数线性拟合,实现对大语言模型理论声明的实证核验。 -- **工作流时序两阶段协同**:在步骤 4(解法审计阶段)执行理论声明与结构分析,若生成器尚未构建则标记为待实测;在步骤 6(生成器构建)之后以及步骤 10(测试验证)与步骤 11(全量审计)中自动触发多规模实测与经验拟合,解决时序依赖冲突。 -- **接口兼容与客观门禁**:保持 22 个 MCP 工具签名与现有对外接口不变,在门禁中提供包含多规模实测数据的结构化证据,杜绝基于猜测的误报。 - ---- - -## 二、系统架构与处理流程 - -```mermaid -flowchart TD - A["大语言模型 (Host LLM)"] -->|"推导并声明理论复杂度 (claimed_complexity)"| B["solution_audit_std / solution_analyze"] - B --> C{"生成器 files/gen 是否就绪"} - C -->|"尚未构建 (步骤 4)"| D["记录理论声明,标记待实测"] - C -->|"已构建 (步骤 6 之后 / 步骤 10 / 步骤 11)"| E["多规模阶梯采样器 (MultiScaleSampler)"] - E -->|"自适应计算采样阶梯 (5 点采样)"| F["生成器参数调度与数据生成"] - F --> G["动态执行监测器 (DynamicExecutionMonitor)"] - G -->|"运行纯输入输出基准 (Dummy I/O Runner)"| H["测量环境与启动底噪 T0"] - G -->|"并发运行交互题管道或普通解法"| I["采集纯 CPU 耗时 (utime+stime) 与物理内存"] - I --> J["经验倍率拟合分析器 (EmpiricalRatioAnalyzer)"] - J -->|"对数线性回归与动态倍率核验"| K{"实证比对判定"} - K -->|"拟合吻合且无超时"| L["放行并输出结构化实证证据卡片"] - K -->|"拟合失败或发生超时"| M["阻断并向 next_actions 注入修复建议"] -``` - ---- - -## 三、核心模块详细设计 - -### 1. 多规模阶梯数据采样器(MultiScaleSampler) - -根据大语言模型声明的时间复杂度类别与约束规模 $N_{\max}$,自适应计算采样点数值,避免采样点倒挂: - -1. **多项式时间复杂度($N_{\max} \ge 1000$)**: - 采用 5 点阶梯采样,覆盖不同数量级: - - $N_1 = \max(100, \lfloor N_{\max} \times 0.02 \rfloor)$ - - $N_2 = \max(200, \lfloor N_{\max} \times 0.05 \rfloor)$ - - $N_3 = \max(500, \lfloor N_{\max} \times 0.10 \rfloor)$ - - $N_4 = \max(1000, \lfloor N_{\max} \times 0.30 \rfloor)$ - - $N_5 = N_{\max}$ - 严格保证 $N_1 < N_2 < N_3 < N_4 < N_5 \le N_{\max}$。 - -2. **小规模多项式时间复杂度($N_{\max} < 1000$)**: - 采用等比动态比例缩放: - - $N_1 = \max(10, \lfloor N_{\max} \times 0.20 \rfloor)$ - - $N_2 = \lfloor N_{\max} \times 0.40 \rfloor$ - - $N_3 = \lfloor N_{\max} \times 0.60 \rfloor$ - - $N_4 = \lfloor N_{\max} \times 0.80 \rfloor$ - - $N_5 = N_{\max}$ - -3. **指数与阶乘时间复杂度($N_{\max} \le 30$)**: - 采用增量线性阶梯,杜绝指数爆炸: - - $N_1 = \max(4, N_{\max} - 4)$ - - $N_2 = N_{\max} - 3$ - - $N_3 = N_{\max} - 2$ - - $N_4 = N_{\max} - 1$ - - $N_5 = N_{\max}$ - -4. **对数与常数时间复杂度($O(\log N)$ / $O(1)$)**: - 在基准点 $N_1$ 与极限点 $N_5$ 执行物理运行,验证绝对耗时处于极小常数区间。 - -### 2. 动态执行监测器(DynamicExecutionMonitor) - -1. **纯 CPU 时间采集**: - - 在 Linux 操作系统中,调用 `getrusage(RUSAGE_CHILDREN)` 或直接读取 `/proc/[pid]/stat` 获取子进程的 `utime`(用户态时间)与 `stime`(内核态时间),彻底剥离操作系统进程派生、动态库加载以及管道初始化的物理挂钟时间。 - - 在 Windows 操作系统中,在子进程退出后调用 `GetProcessTimes` 函数分别获取目标进程的 `UserTime` 与 `KernelTime` 之和。在评测期间调用 `timeBeginPeriod(1)` 提升定时器分辨率至 1 毫秒,评测完成后调用 `timeEndPeriod(1)` 还原。 - -2. **环境底噪扣除(Dummy I/O Runner)**: - 在执行待测算法前,先行运行仅包含相同输入读取逻辑的空操作基准程序,测量纯输入输出与环境初始化耗时 $T_0$。在计算算法实际耗时时执行 $T_{\text{calc}} = \max(0.1, T_{\text{measured}} - T_0)$,消除小规模测点下的环境底噪压缩,杜绝小常数二次方算法被误判放行。 - -3. **短耗时自适应提升**: - 当基准点纯计算耗时低于 5 毫秒的安全信噪比门限时,自适应向上调整低阶规模参数,确保采样点处于高信噪比区间。 - -4. **进程与资源安全控制**: - - Linux 环境下创建子进程时开启 `start_new_session=True`,发生超时或异常时向进程组发送 `SIGKILL` 信号(`os.killpg`)进行原子化清理。 - - Windows 环境下为子进程关联 Job Object 并配置 `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE` 标志,由操作系统内核执行深层回收。 - - 标准输入输出采用异步非阻塞读取机制,重定向至带大小上限的缓冲区,避免缓冲区填满引发主控与子进程相互挂起。 - -5. **交互题双向管道调度管线**: - 当 `manifest.json` 声明 `interactive: true` 时,并发启动 `files/interactor` 与 `solutions/sol`,通过匿名管道互联两者的标准输入输出,独立采集 `sol` 进程的 CPU 时间,排除交互器自身的计算与调度干扰。 - -### 3. 经验倍率拟合分析器(EmpiricalRatioAnalyzer) - -1. **动态理论倍率计算模型**: - 废除固定的 10 倍倍率表,根据实际采样的具体规模数值 $N_a$ 与 $N_b$($N_a < N_b$),动态计算理论期望倍率: - $$R_{\text{expected}} = \frac{f(N_b)}{f(N_a)}$$ - 实际观测倍率为 $R_{\text{observed}} = \frac{T_{\text{calc}}(N_b)}{T_{\text{calc}}(N_a)}$。 - 容差区间动态调节: - $$R_{\text{observed}} \in \left[ R_{\text{expected}} \times (1 - \delta), R_{\text{expected}} \times (1 + \delta) \right]$$ - 容差系数 $\delta$ 依据基准点耗时自适应设定。 - -2. **对数线性回归拟合模型**: - 通过 5 个阶梯采样点拟合经验幂函数模型: - $$\ln(T_{\text{calc}}(N)) = \alpha \ln N + \beta$$ - 对拟合所得的幂指数 $\alpha$ 进行区间核验: - - 线性复杂度 $O(N)$:$\alpha \in [0.8, 1.3]$ - - 线性对数复杂度 $O(N \log N)$:$\alpha \in [1.0, 1.4]$ - - 二次方复杂度 $O(N^2)$:$\alpha \in [1.7, 2.3]$ - - 三次方复杂度 $O(N^3)$:$\alpha \in [2.6, 3.4]$ - 结合判定系数 $R^2 \ge 0.85$ 评估拟合优度,消除大常数项对增长率的稀释影响。 - -3. **处理器缓存容量跨越保护**: - 实施双区间联合校验。若仅在极限点 $N_5$ 发生微幅跳跃,且极限点绝对耗时处于时限的安全比例内(例如低于时限的 50%),判定为硬件缓存效应并予以放行,记录分析说明。 - ---- - -## 四、接口契约与数据结构规范 - -### 1. 输入参数归一化与轻量化 - -大语言模型调用 `solution_analyze` 与 `solution_audit_std` 时,`claimed_complexity` 为可选参数(`Optional[str] = None`)。 -工具端实施输入归一化解析: -- 自动消除空白字符、转换为小写字符。 -- 自动去除 LaTeX 转义符(例如 `\log` 转换为 `log`,`\cdot` 转换为 `*`)。 -- 规模变量范围与时限自动从 `.autocode/manifest.json` 中提取,无需大语言模型重复传递。 - -### 2. 结构化返回格式规范 - -#### 验证通过时的返回结构(嵌入 `data["empirical_verification"]`): -```json -{ - "empirical_verification": { - "passed": true, - "verdict": "verified", - "claimed_complexity": "O(n log n)", - "fitted_complexity": "O(n log n)", - "fitted_alpha": 1.12, - "r_squared": 0.985, - "samples": [ - {"n": 2000, "cpu_time_ms": 2.4, "memory_mb": 3.1, "status": "ok"}, - {"n": 5000, "cpu_time_ms": 6.8, "memory_mb": 4.2, "status": "ok"}, - {"n": 10000, "cpu_time_ms": 14.5, "memory_mb": 6.0, "status": "ok"}, - {"n": 30000, "cpu_time_ms": 48.2, "memory_mb": 11.5, "status": "ok"}, - {"n": 100000, "cpu_time_ms": 175.0, "memory_mb": 24.8, "status": "ok"} - ], - "growth_ratios": [ - {"from_n": 10000, "to_n": 100000, "scale_factor": 10.0, "observed_ratio": 12.07, "expected_ratio": 12.5} - ], - "max_scale_headroom_ratio": 0.175 - } -} -``` - -#### 验证失败阻断时的返回结构: -```json -{ - "empirical_verification": { - "passed": false, - "verdict": "ratio_mismatch", - "failure_reason": "Measured growth ratio 92.5x significantly exceeds theoretical expectation [10.0x, 15.0x] for O(n log n)", - "claimed_complexity": "O(n log n)", - "fitted_complexity": "O(n^2)", - "fitted_alpha": 2.04, - "r_squared": 0.991, - "samples": [ - {"n": 2000, "cpu_time_ms": 2.1, "memory_mb": 3.0, "status": "ok"}, - {"n": 5000, "cpu_time_ms": 13.5, "memory_mb": 4.1, "status": "ok"}, - {"n": 10000, "cpu_time_ms": 55.2, "memory_mb": 6.0, "status": "ok"}, - {"n": 30000, "cpu_time_ms": 502.1, "memory_mb": 11.2, "status": "ok"}, - {"n": 100000, "cpu_time_ms": null, "memory_mb": null, "status": "timeout"} - ], - "remediation_advice": "The implementation exhibits quadratic growth O(n^2). Please inspect nested loops, optimize algorithm logic to O(n log n), or update claimed_complexity in manifest if quadratic complexity is intended." - } -} -``` - -在 `problem_audit` 中,若经验拟合未通过,向 `blocking_issues` 追加条目,并在 `next_actions` 列表中生成优先级为 `high` 的优化建议动作。 - ---- - -## 五、测试套件详细设计 - -### 1. 单元测试(Unit Tests) -- `tests/test_tools/test_empirical_ratio_analyzer.py`: - - 典型复杂度曲线拟合测试($O(N), O(N \log N), O(N^2), O(N^3)$ 样本)。 - - 动态倍率期望与容差区间计算测试。 - - 测量波动($\pm 15\%$)抗干扰鲁棒性测试。 - - 逆序耗时样本异常识别测试。 - - 环境底噪扣除稳定性测试。 -- `tests/test_tools/test_multi_scale_sampler.py`: - - 采样点数值计算单调性测试(覆盖多项式、小规模、指数级边界)。 - - 生成器调用参数规范格式化测试。 -- `tests/test_utils/test_dynamic_execution_monitor.py`: - - 跨平台纯 CPU 耗时采集精度测试(Linux 与 Windows)。 - - 异常超时进程树原子化清理测试。 - -### 2. 端到端集成测试(Integration Tests) -- `tests/test_integration/test_complexity_empirical_e2e.py`: - - 正确线性标答放行测试(单调队列或双指针)。 - - 小常数二次方算法阻断测试(在 $N=10000$ 下被准确识别并阻断)。 - - 交互题双向管道执行与 CPU 时间采集测试。 - - 多测数据极端分布(大 $T$ 小 $N$ 与小 $T$ 大 $N$)评测。 diff --git a/docs/plans/multi-scale-complexity-empirical-fitting-plan.md b/docs/plans/multi-scale-complexity-empirical-fitting-plan.md deleted file mode 100644 index 93d5551..0000000 --- a/docs/plans/multi-scale-complexity-empirical-fitting-plan.md +++ /dev/null @@ -1,79 +0,0 @@ -# 多规模阶梯数据采样与经验复杂度拟合执行计划 - -## 一、概述 -本计划依据 [docs/designs/multi-scale-complexity-empirical-fitting-design.md](file:///home/cvm-204/AutoCode/docs/designs/multi-scale-complexity-empirical-fitting-design.md),实施对 AutoCode 复杂度与算法分析体系的全面重构。 -移除原有的脆弱正则表达式静态分析,建立“大语言模型声明理论复杂度 + 底层工具链多规模物理实测与对数线性倍率拟合”的确定性实证架构。 - ---- - -## 二、任务分解与实施步骤 - -### 任务 1:经验倍率拟合分析器(EmpiricalRatioAnalyzer)与数学单元测试 -- **目标文件**: - - 新建 `src/autocode_mcp/utils/ratio_analyzer.py` - - 新建 `tests/test_tools/test_empirical_ratio_analyzer.py` -- **核心逻辑**: - 1. 动态理论期望倍率计算函数:根据任意给定的两测点规模 $N_a, N_b$ 动态求取 $R_{\text{expected}} = f(N_b) / f(N_a)$。 - 2. 对数线性回归拟合模型:计算 $\ln(T_{\text{calc}}) = \alpha \ln N + \beta$ 的幂指数 $\alpha$ 与判定系数 $R^2$。 - 3. 容差检验与复杂度判定函数:比对实测幂指数 $\alpha$ 与观测倍率,输出结构化判定结果(`verified` 或 `ratio_mismatch`)。 -- **验证命令**: - `uv run pytest tests/test_tools/test_empirical_ratio_analyzer.py -v` - ---- - -### 任务 2:自适应多规模阶梯数据采样器(MultiScaleSampler)与单元测试 -- **目标文件**: - - 新建 `src/autocode_mcp/utils/scale_sampler.py` - - 新建 `tests/test_tools/test_multi_scale_sampler.py` -- **核心逻辑**: - 1. 自适应采样点计算:依据 $N_{\max}$ 与时间复杂度类别自适应计算 5 个单调递增采样点。杜绝 $N_{\max} \le 30$ 时的数值倒挂。 - 2. 生成器调用参数规范化:支持位置参数、命名参数与多测参数注入。 - 3. 多测($T$ 组数据)极端数据分布生成逻辑(大 $T$ 小 $N$ 与小 $T$ 大 $N$)。 -- **验证命令**: - `uv run pytest tests/test_tools/test_multi_scale_sampler.py -v` - ---- - -### 任务 3:纯 CPU 耗时采集与环境底噪扣除监控器(DynamicExecutionMonitor) -- **目标文件**: - - 新建 `src/autocode_mcp/utils/execution_monitor.py` - - 新建 `tests/test_utils/test_execution_monitor.py` -- **核心逻辑**: - 1. 跨平台纯 CPU 耗时采集:Linux 环境使用 `getrusage` / `/proc/[pid]/stat`;Windows 环境调用 `GetProcessTimes` 并配合 `timeBeginPeriod(1)`。 - 2. 空操作基准程序(Dummy I/O Runner)测量环境底噪与启动耗时 $T_0$,计算纯算法计算耗时 $T_{\text{calc}} = \max(0.1, T_{\text{measured}} - T_0)$。 - 3. 交互题双向匿名管道并发调度与独立 CPU 耗时统计。 - 4. 进程树生命周期管控:Linux 会话隔离与 `os.killpg`;Windows Job Object 内核清理。 -- **验证命令**: - `uv run pytest tests/test_utils/test_execution_monitor.py -v` - ---- - -### 任务 4:工具层整合与脆弱正则清理 -- **目标文件**: - - 修改 `src/autocode_mcp/tools/complexity.py`:清理 `analyze_loop_complexity` 与 `detect_algorithm_patterns` 中的正则表达式猜测试图;接入 `MultiScaleSampler`、`DynamicExecutionMonitor` 与 `EmpiricalRatioAnalyzer`。 - - 修改 `src/autocode_mcp/tools/solution_audit.py`:在 `solution_audit_std` 与 `solution_audit_brute` 中支持 `claimed_complexity` 归一化输入与多规模经验拟合结果嵌入。 - - 修改 `src/autocode_mcp/tools/audit.py`:在全量审计中将经验拟合指标纳入质量信号,失败时向 `blocking_issues` 与 `next_actions` 追加明确指引。 -- **两阶段时序控制**: - 在解法审计(步骤 4)时若 `files/gen` 未生成,标记为待实测;在生成器构建后及步骤 10 与步骤 11 自动执行完整多规模经验验证。 -- **验证命令**: - `uv run pytest tests/test_tools/test_complexity.py tests/test_tools/test_solution_audit.py -v` - ---- - -### 任务 5:端到端集成测试与小常数二次方算法证伪验证 -- **目标文件**: - - 新建 `tests/test_integration/test_complexity_empirical_e2e.py` -- **核心测试用例**: - 1. 线性标答(单调队列)在 $N=10^5$ 规模下通过验证并输出完整实测证据卡片。 - 2. 声明 $O(N)$ 但实现小常数 $O(N^2)$ 的伪装算法在 $N=10^4$ 规模下被准确识别为二次方增长并就地阻断。 - 3. 交互题双向管道执行与纯 CPU 时间采集准确性测试。 - 4. 生成器参数缺失时的极限单点自适应回退测试。 -- **验证命令**: - `uv run pytest tests/test_integration/test_complexity_empirical_e2e.py -v` - ---- - -### 任务 6:代码规范与全量回归测试 -- **静态代码检查**:`uv run ruff check .` -- **类型系统检查**:`uv run mypy src/` -- **全量测试套件执行**:`uv run pytest tests/ -q` diff --git a/src/autocode_mcp/tools/complexity.py b/src/autocode_mcp/tools/complexity.py index e5e1518..9f80606 100644 --- a/src/autocode_mcp/tools/complexity.py +++ b/src/autocode_mcp/tools/complexity.py @@ -203,13 +203,20 @@ async def run_empirical_verification( "message": "files/gen or binary not built yet, empirical verification will execute after generator_build", } + if not claimed_complexity: + return { + "status": "skipped", + "passed": True, + "message": "claimed_complexity not provided; skipping empirical verification to avoid false assumptions", + } + actual_n_max = 10000 time_limit_ms = 2000.0 if constraints: actual_n_max = int(constraints.get("n_max") or 10000) time_limit_ms = float(constraints.get("time_limit_ms") or 2000.0) - effective_complexity = claimed_complexity or "O(n)" + effective_complexity = claimed_complexity scale_points = MultiScaleSampler.compute_scale_points(actual_n_max, effective_complexity) empirical_dir = os.path.join(problem_dir, ".autocode", "empirical_tests") @@ -353,7 +360,7 @@ async def execute( empirical_verification = await run_empirical_verification( problem_dir, solution_type, - claimed_complexity or detected_claimed or final_complexity, + claimed_complexity or detected_claimed, constraints, ) diff --git a/src/autocode_mcp/utils/execution_monitor.py b/src/autocode_mcp/utils/execution_monitor.py index 62df0aa..41cccd1 100644 --- a/src/autocode_mcp/utils/execution_monitor.py +++ b/src/autocode_mcp/utils/execution_monitor.py @@ -98,13 +98,34 @@ async def _sample_resources() -> None: sampler_task = asyncio.create_task(_sample_resources()) + async def _read_stream_limited(reader: asyncio.StreamReader | None, limit: int) -> bytes: + if not reader: + return b"" + chunks: list[bytes] = [] + collected = 0 + while not reader.at_eof(): + try: + chunk = await reader.read(65536) + except (BrokenPipeError, ConnectionResetError, asyncio.CancelledError): + break + if not chunk: + break + if collected < limit: + remaining = limit - collected + chunks.append(chunk[:remaining]) + collected += min(len(chunk), remaining) + return b"".join(chunks) + + read_stdout_task = asyncio.create_task(_read_stream_limited(subproc.stdout, max_output_bytes)) + read_stderr_task = asyncio.create_task(_read_stream_limited(subproc.stderr, max_output_bytes)) + try: - stdout_data, stderr_data = await asyncio.wait_for( - subproc.communicate(), + await asyncio.wait_for( + asyncio.gather(subproc.wait(), read_stdout_task, read_stderr_task), timeout=timeout_sec, ) - stdout_bytes = stdout_data[:max_output_bytes] - stderr_bytes = stderr_data[:max_output_bytes] + stdout_bytes = read_stdout_task.result() + stderr_bytes = read_stderr_task.result() except asyncio.TimeoutError: status = "timeout" await terminate_pid_tree(pid) @@ -112,6 +133,10 @@ async def _sample_resources() -> None: await asyncio.wait_for(subproc.wait(), timeout=1.0) except Exception: pass + if not read_stdout_task.done(): + read_stdout_task.cancel() + if not read_stderr_task.done(): + read_stderr_task.cancel() finally: sampler_task.cancel() try: diff --git a/src/autocode_mcp/utils/ratio_analyzer.py b/src/autocode_mcp/utils/ratio_analyzer.py index 964b543..5f64663 100644 --- a/src/autocode_mcp/utils/ratio_analyzer.py +++ b/src/autocode_mcp/utils/ratio_analyzer.py @@ -68,14 +68,42 @@ def evaluate_complexity_function(cls, normalized_expr: str, n: float) -> float: if normalized_expr in ("O(n^3)",): return n * n * n if normalized_expr in ("O(2^n)",): - return math.pow(2.0, min(n, 60.0)) + return math.pow(2.0, min(n, 1020.0)) if normalized_expr in ("O(n!)",): - return float(math.factorial(min(int(n), 20))) + int_n = max(1, min(int(n), 170)) + return float(math.factorial(int_n)) return n @classmethod def calculate_expected_ratio(cls, claimed_complexity: str, n_from: int, n_to: int) -> float: norm_expr = normalize_complexity_expression(claimed_complexity) + if norm_expr == "O(n!)": + if n_from <= 0: + n_from = 1 + if n_to <= 0: + n_to = 1 + if n_to == n_from: + return 1.0 + if n_to > n_from: + ratio = 1.0 + for k in range(n_from + 1, n_to + 1): + ratio *= k + if ratio > 1e12: + return 1e12 + return ratio + else: + ratio = 1.0 + for k in range(n_to + 1, n_from + 1): + ratio *= k + if ratio > 1e12: + return 0.0 + return 1.0 / ratio if ratio > 0 else 0.0 + + if norm_expr == "O(2^n)": + diff = float(n_to - n_from) + clamped_diff = min(max(diff, -100.0), 100.0) + return math.pow(2.0, clamped_diff) + val_from = cls.evaluate_complexity_function(norm_expr, float(n_from)) val_to = cls.evaluate_complexity_function(norm_expr, float(n_to)) if val_from <= 0: diff --git a/src/autocode_mcp/utils/scale_sampler.py b/src/autocode_mcp/utils/scale_sampler.py index aebe367..ea7edc5 100644 --- a/src/autocode_mcp/utils/scale_sampler.py +++ b/src/autocode_mcp/utils/scale_sampler.py @@ -45,6 +45,7 @@ def format_generator_command( vars_map: dict[str, Any] = { "n": n, "seed": seed, + "type": "random", "n_min": n, "n_max": n, "t_min": 1, @@ -62,9 +63,9 @@ def format_generator_command( return [ generator_exe, str(seed), - "random", + str(vars_map.get("type", "random")), str(vars_map.get("n_min", n)), - str(n), + str(vars_map.get("n_max", n)), str(vars_map.get("t_min", 1)), str(vars_map.get("t_max", 1)), ] diff --git a/tests/test_integration/test_complexity_empirical_e2e.py b/tests/test_integration/test_complexity_empirical_e2e.py index bf455a5..a3e5e9b 100644 --- a/tests/test_integration/test_complexity_empirical_e2e.py +++ b/tests/test_integration/test_complexity_empirical_e2e.py @@ -229,3 +229,27 @@ async def test_generator_missing_fallback_e2e(): assert result.success empirical = result.data.get("empirical_verification", {}) assert empirical.get("status") == "pending_generator" + + +@pytest.mark.asyncio +async def test_solution_without_claimed_complexity_skips_verification(): + tool = SolutionAnalyzeTool() + exe_ext = get_exe_extension() + + with tempfile.TemporaryDirectory() as tmpdir: + gen_bin = os.path.join(tmpdir, "files", f"gen{exe_ext}") + sol_bin = os.path.join(tmpdir, "solutions", f"sol{exe_ext}") + _compile_cpp(GEN_CPP, gen_bin) + _compile_cpp(QUADRATIC_SOL_CPP, sol_bin) + + result = await tool.execute( + problem_dir=tmpdir, + solution_type="sol", + claimed_complexity=None, + constraints={"n_max": 10000, "time_limit_ms": 2000.0}, + ) + + assert result.success + empirical = result.data.get("empirical_verification", {}) + assert empirical.get("status") == "skipped" + assert empirical.get("passed") is True diff --git a/tests/test_tools/test_empirical_ratio_analyzer.py b/tests/test_tools/test_empirical_ratio_analyzer.py index 1a748b7..b667fab 100644 --- a/tests/test_tools/test_empirical_ratio_analyzer.py +++ b/tests/test_tools/test_empirical_ratio_analyzer.py @@ -27,6 +27,12 @@ def test_calculate_expected_ratio() -> None: r_cubic = EmpiricalRatioAnalyzer.calculate_expected_ratio("O(n^3)", 100, 1000) assert pytest.approx(r_cubic, 0.01) == 1000.0 + r_fact = EmpiricalRatioAnalyzer.calculate_expected_ratio("O(n!)", 20, 22) + assert pytest.approx(r_fact, 0.01) == 21.0 * 22.0 + + r_exp = EmpiricalRatioAnalyzer.calculate_expected_ratio("O(2^n)", 60, 65) + assert pytest.approx(r_exp, 0.01) == 32.0 + def test_fit_log_linear_linear_samples() -> None: samples = [ diff --git a/tests/test_tools/test_multi_scale_sampler.py b/tests/test_tools/test_multi_scale_sampler.py index 957ce09..df180c9 100644 --- a/tests/test_tools/test_multi_scale_sampler.py +++ b/tests/test_tools/test_multi_scale_sampler.py @@ -51,6 +51,16 @@ def test_format_generator_command_template() -> None: assert cmd == ["gen.exe", "-n", "5000", "--seed", "42", "-m", "10000"] +def test_format_generator_command_extra_vars_custom_type_and_n_max() -> None: + cmd = MultiScaleSampler.format_generator_command( + "gen.exe", + 5000, + 42, + extra_vars={"type": "2", "n_max": 8000}, + ) + assert cmd == ["gen.exe", "42", "2", "5000", "8000", "1", "1"] + + def test_generate_scale_input_file(tmp_path: os.PathLike[str]) -> None: out_file = os.path.join(str(tmp_path), "test_scale.in") cmd = [sys.executable, "-c", "import sys; print(' '.join(sys.argv[1:]))", "10", "20", "30"] diff --git a/tests/test_utils/test_execution_monitor.py b/tests/test_utils/test_execution_monitor.py index eda459e..754999a 100644 --- a/tests/test_utils/test_execution_monitor.py +++ b/tests/test_utils/test_execution_monitor.py @@ -87,3 +87,23 @@ async def test_run_monitored_process_mle(tmp_path: os.PathLike[str]) -> None: ) assert res["status"] == "mle" + + +@pytest.mark.asyncio +async def test_run_monitored_process_output_limit(tmp_path: os.PathLike[str]) -> None: + in_file = os.path.join(str(tmp_path), "input.in") + with open(in_file, "w", encoding="utf-8") as f: + f.write("1\n") + + code = "import sys\nfor _ in range(200000):\n sys.stdout.write('A' * 64 + '\\n')" + cmd = [sys.executable, "-c", code] + + res = await DynamicExecutionMonitor.run_monitored_process( + cmd, + in_file, + time_limit_ms=5000.0, + memory_limit_mb=512.0, + ) + + assert res["status"] == "ok" + assert len(res["stdout"]) <= 10 * 1024 * 1024