mirror of
https://github.com/val1813/kwcode.git
synced 2026-09-03 06:34:30 +08:00
feat: audit log + model cmd + indent alignment fix
Audit log (audit/logger.py): - Persists task execution trace as human-readable JSON - Storage: ~/.kaiwu/logs/, max 100, auto-cleanup - Records: task, gate, experts, files, tests, retries (no code content) - orchestrator._emit() now instance method, auto-logs to audit - CLI: kwcode log / log show <id> / log clear Model commands (cli/commands/model_cmd.py): - kwcode model: show current config + tier - kwcode model set <name>: switch model (writes config.yaml) - kwcode model probe: detect family/params/quant/reasoning via Ollama API Indent alignment fix (Generator._align_indentation): - Fixes systematic bug: LLM returns class methods at 0-indent, apply_patch replaces 4-indent original → method escapes class - Aligns modified base indent to match original before apply_patch Tests: 10 new (audit 4 + indent 5 + hashline prompt 1), 50 total green Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Val-sss
18
STATUS.md
18
STATUS.md
@@ -53,6 +53,24 @@
|
||||
- orchestrator._record_success/_record_failure_result: 调_record_flywheel()
|
||||
- _record_flywheel(): 策略统计 + 用户模式 + 遥测，三路全非阻塞
|
||||
|
||||
**审计日志** (`audit/logger.py`)
|
||||
- AuditLogger: start() → log(stage, detail) → write(ctx, elapsed, success, model)
|
||||
- 存储：~/.kaiwu/logs/YYYY-MM-DD_HHMMSS_<expert_type>.json
|
||||
- 不记录代码内容，只记录：任务描述/Gate分类/专家执行时间/文件名/测试结果/重试次数
|
||||
- 最多保留100条，超出自动清理
|
||||
- orchestrator._emit()从@staticmethod改为实例方法，每个事件自动记录到audit
|
||||
- CLI: `kwcode log` / `kwcode log show <id>` / `kwcode log clear`
|
||||
|
||||
**kwcode model命令** (`cli/commands/model_cmd.py`)
|
||||
- `kwcode model` — 显示当前模型配置+能力tier
|
||||
- `kwcode model set <名称>` — 切换模型（写入config.yaml）
|
||||
- `kwcode model probe` — 探测模型详情（Ollama API: family/参数量/量化/reasoning）
|
||||
|
||||
**缩进对齐修复** (`Generator._align_indentation`)
|
||||
- 修复系统性bug：LLM返回class方法时丢失缩进（0空格 vs 原始4空格）
|
||||
- apply_patch替换后方法"跑出"class导致IndentationError
|
||||
- 修法：_generate_modified()返回后立刻调_align_indentation()补齐缩进差
|
||||
|
||||
**P0: Hashline锚点编辑** (`tools/hashline.py`)
|
||||
- add_anchors(): 每行加6字符MD5哈希锚点 `行号|哈希| 内容`
|
||||
- parse_anchor_edits(): 解析 EDIT/DELETE/INSERT_AFTER 指令
|
||||
|
||||
0
kaiwu/audit/__init__.py
Normal file
0
kaiwu/audit/__init__.py
Normal file
180
kaiwu/audit/logger.py
Normal file
180
kaiwu/audit/logger.py
Normal file
@@ -0,0 +1,180 @@
|
||||
"""
|
||||
审计日志：持久化任务执行轨迹为人类可读格式。
|
||||
|
||||
存储位置：.kaiwu/logs/YYYY-MM-DD_HHMMSS_<expert_type>.json
|
||||
不记录代码内容，只记录元数据和行为轨迹。
|
||||
最多保留100条，超出自动清理最旧的。
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
LOGS_DIR = Path.home() / ".kaiwu" / "logs"
|
||||
MAX_LOGS = 100
|
||||
|
||||
|
||||
class AuditLogger:
|
||||
"""任务执行审计日志。"""
|
||||
|
||||
def __init__(self):
|
||||
self._events: list[dict] = []
|
||||
self._start_time: float = 0
|
||||
|
||||
def start(self):
|
||||
"""任务开始时调用。"""
|
||||
self._events = []
|
||||
self._start_time = time.time()
|
||||
|
||||
def log(self, stage: str, detail: str):
|
||||
"""记录一个执行事件。"""
|
||||
elapsed = time.time() - self._start_time if self._start_time else 0
|
||||
self._events.append({
|
||||
"time": datetime.now().strftime("%H:%M:%S"),
|
||||
"elapsed_s": round(elapsed, 1),
|
||||
"stage": stage,
|
||||
"detail": detail,
|
||||
})
|
||||
|
||||
def write(self, ctx, elapsed: float, success: bool, model: str = "unknown"):
|
||||
"""
|
||||
任务完成时写入日志文件。非阻塞，失败静默。
|
||||
|
||||
Args:
|
||||
ctx: TaskContext
|
||||
elapsed: 总耗时秒数
|
||||
success: 是否成功
|
||||
model: 模型名称
|
||||
"""
|
||||
try:
|
||||
LOGS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
gate = ctx.gate_result or {}
|
||||
expert_type = gate.get("expert_type", "unknown")
|
||||
difficulty = gate.get("difficulty", "?")
|
||||
ts = datetime.now().strftime("%Y-%m-%d_%H%M%S")
|
||||
filename = f"{ts}_{expert_type}.json"
|
||||
|
||||
# 提取修改的文件列表（不含内容）
|
||||
files_modified = []
|
||||
patches = []
|
||||
if ctx.generator_output:
|
||||
patches = ctx.generator_output.get("patches", [])
|
||||
files_modified = [p.get("file", "") for p in patches]
|
||||
|
||||
# 计算改动行数
|
||||
lines_added = 0
|
||||
lines_removed = 0
|
||||
for p in patches:
|
||||
orig_lines = len(p.get("original", "").split("\n")) if p.get("original") else 0
|
||||
mod_lines = len(p.get("modified", "").split("\n")) if p.get("modified") else 0
|
||||
lines_added += max(0, mod_lines - orig_lines)
|
||||
lines_removed += max(0, orig_lines - mod_lines)
|
||||
|
||||
# 测试结果
|
||||
tests_passed = 0
|
||||
tests_total = 0
|
||||
if ctx.verifier_output:
|
||||
tests_passed = ctx.verifier_output.get("tests_passed", 0)
|
||||
tests_total = ctx.verifier_output.get("tests_total", 0)
|
||||
|
||||
record = {
|
||||
"task": ctx.user_input[:200], # 保留任务描述（用户自己的输入）
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"model": model,
|
||||
"expert_type": expert_type,
|
||||
"difficulty": difficulty,
|
||||
"elapsed_s": round(elapsed, 1),
|
||||
"success": success,
|
||||
"retry_count": ctx.retry_count,
|
||||
"files_modified": files_modified,
|
||||
"lines_added": lines_added,
|
||||
"lines_removed": lines_removed,
|
||||
"tests_passed": tests_passed,
|
||||
"tests_total": tests_total,
|
||||
"search_triggered": ctx.search_triggered,
|
||||
"events": self._events,
|
||||
}
|
||||
|
||||
log_path = LOGS_DIR / filename
|
||||
log_path.write_text(
|
||||
json.dumps(record, ensure_ascii=False, indent=2),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
# 清理超过MAX_LOGS的旧日志
|
||||
self._cleanup()
|
||||
|
||||
except Exception as e:
|
||||
logger.debug("Audit log write failed (non-blocking): %s", e)
|
||||
|
||||
def _cleanup(self):
|
||||
"""保留最近MAX_LOGS条日志，删除最旧的。"""
|
||||
try:
|
||||
logs = sorted(LOGS_DIR.glob("*.json"), key=lambda p: p.name)
|
||||
if len(logs) > MAX_LOGS:
|
||||
for old in logs[:len(logs) - MAX_LOGS]:
|
||||
old.unlink()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def list_logs(limit: int = 20) -> list[dict]:
|
||||
"""列出最近的日志摘要。"""
|
||||
if not LOGS_DIR.exists():
|
||||
return []
|
||||
|
||||
logs = sorted(LOGS_DIR.glob("*.json"), key=lambda p: p.name, reverse=True)
|
||||
result = []
|
||||
for i, path in enumerate(logs[:limit]):
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
result.append({
|
||||
"id": len(logs) - i,
|
||||
"file": path.name,
|
||||
"task": data.get("task", "")[:60],
|
||||
"success": data.get("success", False),
|
||||
"elapsed_s": data.get("elapsed_s", 0),
|
||||
"timestamp": data.get("timestamp", ""),
|
||||
"model": data.get("model", ""),
|
||||
})
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def show_log(log_id: int) -> Optional[dict]:
|
||||
"""获取指定ID的日志详情。"""
|
||||
if not LOGS_DIR.exists():
|
||||
return None
|
||||
|
||||
logs = sorted(LOGS_DIR.glob("*.json"), key=lambda p: p.name)
|
||||
idx = log_id - 1
|
||||
if idx < 0 or idx >= len(logs):
|
||||
return None
|
||||
|
||||
try:
|
||||
return json.loads(logs[idx].read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def clear_logs() -> int:
|
||||
"""清除所有日志，返回删除数量。"""
|
||||
if not LOGS_DIR.exists():
|
||||
return 0
|
||||
count = 0
|
||||
for path in LOGS_DIR.glob("*.json"):
|
||||
try:
|
||||
path.unlink()
|
||||
count += 1
|
||||
except Exception:
|
||||
pass
|
||||
return count
|
||||
108
kaiwu/cli/commands/log_cmd.py
Normal file
108
kaiwu/cli/commands/log_cmd.py
Normal file
@@ -0,0 +1,108 @@
|
||||
"""
|
||||
kwcode log — 任务历史审计日志命令。
|
||||
"""
|
||||
|
||||
import typer
|
||||
from rich.prompt import Confirm
|
||||
|
||||
from kaiwu.cli.formatters import console
|
||||
|
||||
log_app = typer.Typer(name="log", help="任务历史日志")
|
||||
|
||||
|
||||
@log_app.callback(invoke_without_command=True)
|
||||
def log_list(
|
||||
ctx: typer.Context,
|
||||
limit: int = typer.Option(20, "--limit", "-n", help="显示条数"),
|
||||
):
|
||||
"""查看最近的任务历史。"""
|
||||
if ctx.invoked_subcommand is not None:
|
||||
return
|
||||
|
||||
from kaiwu.audit.logger import list_logs
|
||||
|
||||
logs = list_logs(limit=limit)
|
||||
if not logs:
|
||||
console.print(" [dim]暂无任务记录[/dim]")
|
||||
return
|
||||
|
||||
console.print()
|
||||
console.print(" [bold]任务历史[/bold]")
|
||||
console.print(" " + "─" * 55)
|
||||
|
||||
for entry in logs:
|
||||
icon = "[green]✅[/green]" if entry["success"] else "[red]❌[/red]"
|
||||
elapsed = f"{entry['elapsed_s']:.0f}s"
|
||||
task = entry["task"][:40]
|
||||
ts = entry["timestamp"][:16].replace("T", " ") if entry["timestamp"] else ""
|
||||
|
||||
console.print(
|
||||
f" #{entry['id']:<3} {icon} {elapsed:>4} {task:<40} [dim]{ts}[/dim]"
|
||||
)
|
||||
|
||||
console.print()
|
||||
console.print(" [dim]查看详情：kwcode log show <编号>[/dim]")
|
||||
console.print()
|
||||
|
||||
|
||||
@log_app.command("show")
|
||||
def log_show(
|
||||
log_id: int = typer.Argument(..., help="日志编号"),
|
||||
):
|
||||
"""查看某次任务的详细执行过程。"""
|
||||
from kaiwu.audit.logger import show_log
|
||||
|
||||
data = show_log(log_id)
|
||||
if not data:
|
||||
console.print(f" [red]未找到日志 #{log_id}[/red]")
|
||||
return
|
||||
|
||||
icon = "✅ 成功" if data.get("success") else "❌ 失败"
|
||||
|
||||
console.print()
|
||||
console.print(f" [bold]任务 #{log_id} 详情[/bold]")
|
||||
console.print(" " + "─" * 45)
|
||||
console.print(f" 任务：{data.get('task', '')[:80]}")
|
||||
console.print(f" 时间：{data.get('timestamp', '')[:19]}")
|
||||
console.print(f" 模型：{data.get('model', '')}")
|
||||
console.print(f" 耗时：{data.get('elapsed_s', 0):.1f}秒")
|
||||
console.print(f" 结果：{icon}")
|
||||
console.print()
|
||||
|
||||
# 执行过程
|
||||
events = data.get("events", [])
|
||||
if events:
|
||||
console.print(" [bold cyan]执行过程[/bold cyan]")
|
||||
for ev in events:
|
||||
stage = ev.get("stage", "")
|
||||
detail = ev.get("detail", "")[:80]
|
||||
t = ev.get("time", "")
|
||||
console.print(f" {t} [{stage:<15}] {detail}")
|
||||
console.print()
|
||||
|
||||
# 修改文件
|
||||
files = data.get("files_modified", [])
|
||||
if files:
|
||||
console.print(f" 修改文件：{', '.join(files)}")
|
||||
console.print(f" 改动规模：+{data.get('lines_added', 0)}行 -{data.get('lines_removed', 0)}行")
|
||||
|
||||
tp = data.get("tests_passed", 0)
|
||||
tt = data.get("tests_total", 0)
|
||||
if tt > 0:
|
||||
console.print(f" 测试结果：{tp}/{tt}")
|
||||
|
||||
console.print(f" 重试次数：{data.get('retry_count', 0)}")
|
||||
console.print()
|
||||
|
||||
|
||||
@log_app.command("clear")
|
||||
def log_clear():
|
||||
"""清除所有任务日志。"""
|
||||
confirm = Confirm.ask(" 确认清除所有任务日志？", default=False)
|
||||
if not confirm:
|
||||
console.print(" [dim]已取消[/dim]")
|
||||
return
|
||||
|
||||
from kaiwu.audit.logger import clear_logs
|
||||
count = clear_logs()
|
||||
console.print(f" [green]已清除 {count} 条日志[/green]")
|
||||
118
kaiwu/cli/commands/model_cmd.py
Normal file
118
kaiwu/cli/commands/model_cmd.py
Normal file
@@ -0,0 +1,118 @@
|
||||
"""
|
||||
kwcode model — 模型查看/切换/探测命令。
|
||||
"""
|
||||
|
||||
import typer
|
||||
|
||||
from kaiwu.cli.formatters import console
|
||||
|
||||
model_app = typer.Typer(name="model", help="模型管理")
|
||||
|
||||
|
||||
@model_app.callback(invoke_without_command=True)
|
||||
def model_show(ctx: typer.Context):
|
||||
"""查看当前模型配置。"""
|
||||
if ctx.invoked_subcommand is not None:
|
||||
return
|
||||
|
||||
from kaiwu.cli.onboarding import load_config
|
||||
|
||||
config = load_config()
|
||||
cfg = config.get("default", {})
|
||||
model = cfg.get("model", "未配置")
|
||||
base_url = cfg.get("base_url", "http://localhost:11434")
|
||||
has_key = bool(cfg.get("api_key"))
|
||||
|
||||
console.print()
|
||||
console.print(f" [bold]当前模型配置[/bold]")
|
||||
console.print(" " + "─" * 40)
|
||||
console.print(f" 模型：{model}")
|
||||
console.print(f" API ：{base_url}")
|
||||
console.print(f" Key ：{'已配置' if has_key else '（无）'}")
|
||||
|
||||
# Try detect tier
|
||||
try:
|
||||
from kaiwu.core.model_capability import detect_model_tier
|
||||
tier = detect_model_tier(model, base_url)
|
||||
console.print(f" 能力：{tier.value}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
console.print()
|
||||
console.print(" [dim]切换模型：kwcode model set <模型名>[/dim]")
|
||||
console.print(" [dim]探测能力：kwcode model probe[/dim]")
|
||||
console.print()
|
||||
|
||||
|
||||
@model_app.command("set")
|
||||
def model_set(
|
||||
name: str = typer.Argument(..., help="模型名称"),
|
||||
):
|
||||
"""切换模型（写入 config.yaml）。"""
|
||||
from kaiwu.cli.onboarding import load_config, _save_config
|
||||
|
||||
config = load_config()
|
||||
config.setdefault("default", {})
|
||||
config["default"]["model"] = name
|
||||
_save_config(config)
|
||||
console.print(f" [green]✓ 模型已切换为 {name}[/green]")
|
||||
console.print(" [dim]重新启动 kwcode 生效[/dim]")
|
||||
|
||||
|
||||
@model_app.command("probe")
|
||||
def model_probe():
|
||||
"""探测当前模型能力（参数量/上下文/推理能力）。"""
|
||||
from kaiwu.cli.onboarding import load_config
|
||||
|
||||
config = load_config()
|
||||
cfg = config.get("default", {})
|
||||
model = cfg.get("model", "未配置")
|
||||
base_url = cfg.get("base_url", "http://localhost:11434")
|
||||
|
||||
console.print(f" 探测模型 {model}...")
|
||||
|
||||
# Try Ollama API
|
||||
try:
|
||||
import httpx
|
||||
resp = httpx.post(
|
||||
f"{base_url}/api/show",
|
||||
json={"name": model},
|
||||
timeout=10,
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
params = data.get("details", {}).get("parameter_size", "未知")
|
||||
family = data.get("details", {}).get("family", "未知")
|
||||
quant = data.get("details", {}).get("quantization_level", "未知")
|
||||
console.print()
|
||||
console.print(f" [bold]模型信息[/bold]")
|
||||
console.print(" " + "─" * 40)
|
||||
console.print(f" 名称：{model}")
|
||||
console.print(f" 家族：{family}")
|
||||
console.print(f" 参数：{params}")
|
||||
console.print(f" 量化：{quant}")
|
||||
|
||||
# Detect reasoning
|
||||
try:
|
||||
from kaiwu.llm.llama_backend import LLMBackend
|
||||
is_reasoning = LLMBackend._check_reasoning_model(model)
|
||||
console.print(f" 推理：{'✅ reasoning模型' if is_reasoning else '标准模型'}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Detect tier
|
||||
try:
|
||||
from kaiwu.core.model_capability import detect_model_tier
|
||||
tier = detect_model_tier(model, base_url)
|
||||
console.print(f" 能力：{tier.value}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
console.print()
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
console.print(f" [yellow]无法通过 Ollama API 探测 {model}[/yellow]")
|
||||
console.print(f" [dim]请确认 {base_url} 可访问且模型已下载[/dim]")
|
||||
console.print()
|
||||
@@ -55,6 +55,8 @@ from kaiwu.cli.commands.task import (
|
||||
build_pipeline,
|
||||
run_task,
|
||||
)
|
||||
from kaiwu.cli.commands.log_cmd import log_app
|
||||
from kaiwu.cli.commands.model_cmd import model_app
|
||||
from kaiwu.cli.repl import repl
|
||||
|
||||
app = typer.Typer(
|
||||
@@ -68,6 +70,8 @@ app.add_typer(expert_app)
|
||||
app.add_typer(checkpoint_app)
|
||||
app.add_typer(telemetry_app)
|
||||
app.add_typer(skill_app)
|
||||
app.add_typer(log_app)
|
||||
app.add_typer(model_app)
|
||||
app.command("init")(cmd_init)
|
||||
app.command("memory")(cmd_memory)
|
||||
app.command("status")(cmd_status)
|
||||
|
||||
@@ -40,6 +40,7 @@ from kaiwu.notification.flywheel_notifier import FlywheelNotifier
|
||||
from kaiwu.flywheel.strategy_stats import StrategyStats
|
||||
from kaiwu.flywheel.user_pattern_memory import UserPatternMemory
|
||||
from kaiwu.telemetry.client import TelemetryClient
|
||||
from kaiwu.audit.logger import AuditLogger
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -133,6 +134,7 @@ class PipelineOrchestrator:
|
||||
self._strategy_stats = StrategyStats()
|
||||
self._user_patterns = UserPatternMemory()
|
||||
self._telemetry = TelemetryClient()
|
||||
self._audit = AuditLogger()
|
||||
self.bus = bus or EventBus()
|
||||
self._wink = WinkMonitor()
|
||||
self._cognitive_gate = CognitiveGate()
|
||||
@@ -156,6 +158,7 @@ class PipelineOrchestrator:
|
||||
Returns {"success": bool, "context": TaskContext, "error": str|None, "elapsed": float}.
|
||||
"""
|
||||
start_time = time.time()
|
||||
self._audit.start()
|
||||
|
||||
# 任务级超时看门狗
|
||||
TASK_TIMEOUT_S = 300 # 单任务最长5分钟
|
||||
@@ -573,6 +576,8 @@ class PipelineOrchestrator:
|
||||
self._persist_reflection(project_root, ctx, gate_result, success=True)
|
||||
# 飞轮：策略统计 + 用户模式 + 遥测
|
||||
self._record_flywheel(ctx, gate_result, True)
|
||||
# 审计日志
|
||||
self._audit.write(ctx, elapsed, True, getattr(self, '_model_name', 'unknown'))
|
||||
return {
|
||||
"success": True,
|
||||
"context": ctx,
|
||||
@@ -611,6 +616,8 @@ class PipelineOrchestrator:
|
||||
self._persist_reflection(project_root, ctx, gate_result, success=False)
|
||||
# 飞轮：策略统计 + 用户模式 + 遥测
|
||||
self._record_flywheel(ctx, gate_result, False)
|
||||
# 审计日志
|
||||
self._audit.write(ctx, elapsed, False, getattr(self, '_model_name', 'unknown'))
|
||||
return {
|
||||
"success": False,
|
||||
"context": ctx,
|
||||
@@ -823,12 +830,12 @@ class PipelineOrchestrator:
|
||||
logger.debug("Reviewer failed (non-blocking): %s", e)
|
||||
return {"aligned": True, "confidence": 0.0, "gap": ""}
|
||||
|
||||
@staticmethod
|
||||
def _emit(callback, stage: str, detail: str):
|
||||
"""Emit status update if callback provided."""
|
||||
def _emit(self, callback, stage: str, detail: str):
|
||||
"""Emit status update if callback provided. Also logs to audit."""
|
||||
if callback:
|
||||
callback(stage, detail)
|
||||
logger.info("[%s] %s", stage, detail)
|
||||
self._audit.log(stage, detail)
|
||||
|
||||
def _get_max_retries(self, gate_result: dict) -> int:
|
||||
"""Dynamic retry budget based on task difficulty."""
|
||||
|
||||
@@ -277,6 +277,9 @@ class GeneratorExpert:
|
||||
if not modified:
|
||||
continue
|
||||
|
||||
# 对齐缩进：LLM返回的modified可能丢失class内方法的缩进
|
||||
modified = self._align_indentation(original, modified)
|
||||
|
||||
# Verify original exists in file (should always be true since we read it)
|
||||
if original not in content:
|
||||
logger.error("Extracted original not found in file — this should not happen")
|
||||
@@ -573,6 +576,46 @@ class GeneratorExpert:
|
||||
ctx.generator_output = result
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def _align_indentation(original: str, modified: str) -> str:
|
||||
"""
|
||||
让modified的基础缩进和original保持一致。
|
||||
解决LLM生成class方法时丢失缩进的系统性bug：
|
||||
original有4空格缩进（class内方法），但LLM返回0空格（顶层函数），
|
||||
apply_patch替换后方法"跑出"class，导致IndentationError。
|
||||
"""
|
||||
orig_lines = original.split("\n")
|
||||
mod_lines = modified.split("\n")
|
||||
if not orig_lines or not mod_lines:
|
||||
return modified
|
||||
|
||||
# 获取original第一个非空行的缩进
|
||||
orig_indent = 0
|
||||
for line in orig_lines:
|
||||
if line.strip():
|
||||
orig_indent = len(line) - len(line.lstrip())
|
||||
break
|
||||
|
||||
# 获取modified第一个非空行的缩进
|
||||
mod_indent = 0
|
||||
for line in mod_lines:
|
||||
if line.strip():
|
||||
mod_indent = len(line) - len(line.lstrip())
|
||||
break
|
||||
|
||||
diff = orig_indent - mod_indent
|
||||
if diff <= 0:
|
||||
return modified # modified缩进已经>=original，不需要调整
|
||||
|
||||
pad = " " * diff
|
||||
aligned = []
|
||||
for line in mod_lines:
|
||||
if line.strip(): # 非空行加缩进
|
||||
aligned.append(pad + line)
|
||||
else:
|
||||
aligned.append(line)
|
||||
return "\n".join(aligned)
|
||||
|
||||
@staticmethod
|
||||
def _extract_function(content: str, func_name: str) -> Optional[str]:
|
||||
"""Extract a complete function/method from file content by name."""
|
||||
|
||||
165
tests/test_audit_model.py
Normal file
165
tests/test_audit_model.py
Normal file
@@ -0,0 +1,165 @@
|
||||
"""
|
||||
Tests for audit logger, model commands, and _align_indentation fix.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
class TestAuditLogger(unittest.TestCase):
|
||||
"""Test AuditLogger write/list/show/clear."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
self.logs_dir = Path(self.tmpdir) / "logs"
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.tmpdir)
|
||||
|
||||
@patch("kaiwu.audit.logger.LOGS_DIR")
|
||||
def test_write_creates_log(self, mock_dir):
|
||||
mock_dir.__class__ = type(self.logs_dir)
|
||||
from kaiwu.audit.logger import AuditLogger
|
||||
with patch("kaiwu.audit.logger.LOGS_DIR", self.logs_dir):
|
||||
logger = AuditLogger()
|
||||
logger.start()
|
||||
logger.log("gate", "locator_repair | 难度：easy")
|
||||
logger.log("locator", "读取 test.py")
|
||||
|
||||
# Mock context
|
||||
ctx = MagicMock()
|
||||
ctx.user_input = "修复login函数"
|
||||
ctx.gate_result = {"expert_type": "locator_repair", "difficulty": "easy"}
|
||||
ctx.generator_output = {"patches": [{"file": "test.py", "original": "old", "modified": "new"}]}
|
||||
ctx.verifier_output = {"tests_passed": 3, "tests_total": 3}
|
||||
ctx.retry_count = 0
|
||||
ctx.search_triggered = False
|
||||
|
||||
logger.write(ctx, 5.2, True, "qwen3:8b")
|
||||
|
||||
# Verify log file created
|
||||
logs = list(self.logs_dir.glob("*.json"))
|
||||
assert len(logs) == 1
|
||||
|
||||
data = json.loads(logs[0].read_text(encoding="utf-8"))
|
||||
assert data["task"] == "修复login函数"
|
||||
assert data["success"] is True
|
||||
assert data["model"] == "qwen3:8b"
|
||||
assert len(data["events"]) == 2
|
||||
assert data["files_modified"] == ["test.py"]
|
||||
|
||||
@patch("kaiwu.audit.logger.LOGS_DIR")
|
||||
def test_list_logs(self, mock_dir):
|
||||
from kaiwu.audit.logger import list_logs
|
||||
with patch("kaiwu.audit.logger.LOGS_DIR", self.logs_dir):
|
||||
self.logs_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Write 3 log files with distinct names
|
||||
for i in range(3):
|
||||
record = {"task": f"task {i}", "success": True, "elapsed_s": 1.0,
|
||||
"timestamp": f"2026-05-06T10:00:0{i}", "model": "test"}
|
||||
(self.logs_dir / f"2026-05-06_10000{i}_codegen.json").write_text(
|
||||
json.dumps(record), encoding="utf-8"
|
||||
)
|
||||
|
||||
logs = list_logs(limit=10)
|
||||
assert len(logs) == 3
|
||||
|
||||
@patch("kaiwu.audit.logger.LOGS_DIR")
|
||||
def test_clear_logs(self, mock_dir):
|
||||
from kaiwu.audit.logger import AuditLogger, clear_logs
|
||||
with patch("kaiwu.audit.logger.LOGS_DIR", self.logs_dir):
|
||||
self.logs_dir.mkdir(parents=True, exist_ok=True)
|
||||
(self.logs_dir / "test.json").write_text("{}", encoding="utf-8")
|
||||
count = clear_logs()
|
||||
assert count == 1
|
||||
assert len(list(self.logs_dir.glob("*.json"))) == 0
|
||||
|
||||
@patch("kaiwu.audit.logger.LOGS_DIR")
|
||||
def test_max_logs_cleanup(self, mock_dir):
|
||||
from kaiwu.audit.logger import AuditLogger, MAX_LOGS
|
||||
with patch("kaiwu.audit.logger.LOGS_DIR", self.logs_dir):
|
||||
self.logs_dir.mkdir(parents=True, exist_ok=True)
|
||||
# Create MAX_LOGS + 5 files
|
||||
for i in range(MAX_LOGS + 5):
|
||||
(self.logs_dir / f"2026-01-01_{i:06d}_test.json").write_text("{}", encoding="utf-8")
|
||||
|
||||
al = AuditLogger()
|
||||
al._cleanup()
|
||||
|
||||
remaining = list(self.logs_dir.glob("*.json"))
|
||||
assert len(remaining) == MAX_LOGS
|
||||
|
||||
|
||||
class TestAlignIndentation(unittest.TestCase):
|
||||
"""Test Generator._align_indentation — the class method indentation bug fix."""
|
||||
|
||||
def test_class_method_indent_fix(self):
|
||||
from kaiwu.experts.generator import GeneratorExpert
|
||||
# Original is class method (4-space indent)
|
||||
original = " def login(self, user):\n return True"
|
||||
# LLM returns without class indent
|
||||
modified = "def login(self, user):\n if not user:\n return False\n return True"
|
||||
|
||||
result = GeneratorExpert._align_indentation(original, modified)
|
||||
# Should add 4 spaces to all non-empty lines
|
||||
lines = result.split("\n")
|
||||
assert lines[0] == " def login(self, user):"
|
||||
assert lines[1] == " if not user:"
|
||||
assert lines[2] == " return False"
|
||||
assert lines[3] == " return True"
|
||||
|
||||
def test_no_change_needed(self):
|
||||
from kaiwu.experts.generator import GeneratorExpert
|
||||
original = "def foo():\n return 1"
|
||||
modified = "def foo():\n return 2"
|
||||
result = GeneratorExpert._align_indentation(original, modified)
|
||||
assert result == modified
|
||||
|
||||
def test_already_more_indented(self):
|
||||
from kaiwu.experts.generator import GeneratorExpert
|
||||
original = "def foo():\n return 1"
|
||||
modified = " def foo():\n return 2"
|
||||
result = GeneratorExpert._align_indentation(original, modified)
|
||||
# Should not change — modified already more indented
|
||||
assert result == modified
|
||||
|
||||
def test_empty_lines_preserved(self):
|
||||
from kaiwu.experts.generator import GeneratorExpert
|
||||
original = " def foo():\n\n return 1"
|
||||
modified = "def foo():\n\n return 2"
|
||||
result = GeneratorExpert._align_indentation(original, modified)
|
||||
lines = result.split("\n")
|
||||
assert lines[0] == " def foo():"
|
||||
assert lines[1] == "" # Empty line stays empty
|
||||
assert lines[2] == " return 2"
|
||||
|
||||
def test_8_space_indent(self):
|
||||
from kaiwu.experts.generator import GeneratorExpert
|
||||
# Nested class method (8-space indent)
|
||||
original = " def inner(self):\n pass"
|
||||
modified = "def inner(self):\n return 42"
|
||||
result = GeneratorExpert._align_indentation(original, modified)
|
||||
assert result.startswith(" def inner(self):")
|
||||
assert " return 42" in result
|
||||
|
||||
|
||||
class TestHashlinePrompt(unittest.TestCase):
|
||||
"""Verify HASHLINE_PROMPT exists and has required format markers."""
|
||||
|
||||
def test_prompt_exists(self):
|
||||
from kaiwu.experts.generator import HASHLINE_PROMPT
|
||||
assert "{task_description}" in HASHLINE_PROMPT
|
||||
assert "{anchored_code}" in HASHLINE_PROMPT
|
||||
assert "EDIT" in HASHLINE_PROMPT
|
||||
assert "DELETE" in HASHLINE_PROMPT
|
||||
assert "INSERT_AFTER" in HASHLINE_PROMPT
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user