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/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" 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/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..9f80606 100644 --- a/src/autocode_mcp/tools/complexity.py +++ b/src/autocode_mcp/tools/complexity.py @@ -1,438 +1,428 @@ -""" -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", + } + + 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 + 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, + 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..41cccd1 --- /dev/null +++ b/src/autocode_mcp/utils/execution_monitor.py @@ -0,0 +1,336 @@ +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()) + + 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: + await asyncio.wait_for( + asyncio.gather(subproc.wait(), read_stdout_task, read_stderr_task), + timeout=timeout_sec, + ) + stdout_bytes = read_stdout_task.result() + stderr_bytes = read_stderr_task.result() + except asyncio.TimeoutError: + status = "timeout" + await terminate_pid_tree(pid) + try: + 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: + 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..5f64663 --- /dev/null +++ b/src/autocode_mcp/utils/ratio_analyzer.py @@ -0,0 +1,313 @@ +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.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), + "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, 1020.0)) + if normalized_expr in ("O(n!)",): + 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: + 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]: + 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..ea7edc5 --- /dev/null +++ b/src/autocode_mcp/utils/scale_sampler.py @@ -0,0 +1,94 @@ +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, + "type": "random", + "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), + str(vars_map.get("type", "random")), + str(vars_map.get("n_min", n)), + str(vars_map.get("n_max", 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..a3e5e9b --- /dev/null +++ b/tests/test_integration/test_complexity_empirical_e2e.py @@ -0,0 +1,255 @@ +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]) + (a[j] ^ i) + ((cnt + j) & 255); + } + } + 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.25 <= 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" + + +@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_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(): 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..b667fab --- /dev/null +++ b/tests/test_tools/test_empirical_ratio_analyzer.py @@ -0,0 +1,135 @@ +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 + + 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 = [ + {"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..df180c9 --- /dev/null +++ b/tests/test_tools/test_multi_scale_sampler.py @@ -0,0 +1,80 @@ +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_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"] + 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..754999a --- /dev/null +++ b/tests/test_utils/test_execution_monitor.py @@ -0,0 +1,109 @@ +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" + + +@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 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" },