Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions agents/ascend-kernel-developer.md
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,39 @@ Phase 6: 全量用例验证
Phase 7: Trace 记录 (trace-recorder)
```

## Hook 机制说明

本项目的 `.claude/settings.json` 已配置 **toolUse hook**,用于拦截 agent 对 skill 相关脚本的 Bash 调用。

### 被拦截的脚本

| 类别 | 脚本 | 说明 |
|------|------|------|
| 退化检测 | `validate_tilelang_impl.py` | TileLang AST 退化检测 |
| 退化检测 | `validate_ascendc_impl.py` | AscendC AST 退化检测 |
| 评测脚本 | `evaluate_tilelang.sh` | TileLang 功能验证 |
| 评测脚本 | `evaluate_ascendc.sh` | AscendC 功能验证 |
| 构建脚本 | `utils/build_ascendc.py` | AscendC kernel 编译 |
| 验证脚本 | `utils/verification_ascendc.py` | AscendC 正确性验证 |
| 验证脚本 | `utils/verification_tilelang.py` | TileLang 正确性验证 |
| 性能测试 | `performance.py` | 性能对比测试 |
| 批处理 | `batch_run_performance.sh` | 批量性能测试 |

### Hook 行为

1. **拦截**: 当 agent 通过 Bash tool 调用上述脚本时,hook 自动拦截
2. **替换执行**: 由 `.claude/hooks/skill_script_hook.py` 接管执行
3. **等待完成**: hook 等待脚本实际执行完毕(同步阻塞)
4. **返回结果**: 将 exit code、stdout、stderr 以 JSON 格式返回给 agent
5. **Agent 继续**: agent 收到结果后才继续下一步

### 配置位置

- Hook 脚本: `.claude/hooks/skill_script_hook.py`
- Hook 配置: `.claude/settings.json`

> **注意**: 非拦截命令(如普通 `ls`、`cp`、`python` 调用其他脚本)会透传执行,不受影响。

### 退化检测脚本

| 阶段 | 脚本路径 | 说明 |
Expand Down
9 changes: 5 additions & 4 deletions benchmarks/NPUKernelBench/level1/11_GroupNorm.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import torch.nn as nn
import json
import os
import random

class Model(nn.Module):
"""
Expand Down Expand Up @@ -34,13 +35,13 @@ def get_input_groups():
input_groups = []
for idx, case in enumerate(cases):
inputs = case["inputs"]

dtype_map = {
"float32": torch.float32,
"float16": torch.float16,
"bfloat16": torch.bfloat16,
}

x_info = inputs[0]
dtype = dtype_map[x_info["dtype"]]
if idx % 2 == 0:
Expand All @@ -53,15 +54,15 @@ def get_input_groups():
num_groups = None
weight = None
bias = None

for inp in inputs[1:]:
if inp["name"] == "num_groups":
num_groups = inp["value"]
elif inp["name"] == "weight":
weight = torch.randn(inp["shape"], dtype=dtype)
elif inp["name"] == "bias":
bias = torch.randn(inp["shape"], dtype=dtype)

input_groups.append([x, num_groups, weight, bias])
return input_groups

Expand Down
10 changes: 5 additions & 5 deletions benchmarks/NPUKernelBench/level1/25_NLLLoss.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,13 +33,13 @@ def get_input_groups():
json_path = os.path.join(os.path.dirname(__file__), "25_NLLLoss.json")
with open(json_path, "r") as f:
cases = [json.loads(line) for line in f if line.strip()]

input_groups = []
for idx, case in enumerate(cases):
inputs = case["inputs"]
input_info = inputs[0]
target_info = inputs[1]

dtype_map = {
"float32": torch.float32,
"float16": torch.float16,
Expand All @@ -56,19 +56,19 @@ def get_input_groups():

target_range = target_info.get("range", [0, input_info["shape"][1] - 1])
target = torch.randint(target_range[0], target_range[1] + 1, tuple(target_info["shape"]), dtype=torch.int64)

weight = None
ignore_index = -100
reduction = "mean"

for inp in inputs[2:]:
if inp["name"] == "weight":
weight = torch.randn(inp["shape"], dtype=dtype)
elif inp["name"] == "ignore_index":
ignore_index = inp["value"]
elif inp["name"] == "reduction":
reduction = inp["value"]

input_groups.append([input_tensor, target, weight, ignore_index, reduction])
return input_groups

Expand Down
1 change: 1 addition & 0 deletions benchmarks/NPUKernelBench/level2/2_GroupNormSwish.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import torch_npu
import json
import os
import random

class Model(nn.Module):
"""
Expand Down
204 changes: 204 additions & 0 deletions hooks/skill_script_hook.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,204 @@
#!/usr/bin/env python3
"""
Skill 脚本执行 Hook (PreToolUse 协议)。

对 Bash 工具调用进行拦截:
- 命中 INTERCEPTED_PATTERNS: 直接由本 hook 执行脚本,并以 PreToolUse 协议返回
permissionDecision=deny + additionalContext,把执行结果(exit_code/stdout/stderr)
注入回 agent 的上下文,阻止 harness 二次执行。
- 不命中: 输出 permissionDecision=allow,让 harness 正常执行(不在 hook 中重复执行)。

输入:从 stdin 读取 harness PreToolUse JSON:
{"tool_name": "Bash", "tool_input": {"command": "..."}, ...}

输出:单行 PreToolUse JSON 到 stdout。
"""

import json
import os
import re
import subprocess
import sys
import time


# 每个条目: (interpreter_regex, script_basename_regex)
# 只匹配“某个解释器 + 脚本路径”这种真正的脚本调用,
# 不匹配命令字符串里仅出现脚本名的情况(如 cat / grep / echo JSON 等)。
_PY_INTERP = r"(?:python|python3|python3\.\d+)"
_SH_INTERP = r"(?:bash|sh)"

INTERCEPTED_PATTERNS = [
(_PY_INTERP, r"validate_tilelang_impl\.py"),
(_PY_INTERP, r"validate_ascendc_impl\.py"),
(_SH_INTERP, r"evaluate_tilelang\.sh"),
(_SH_INTERP, r"evaluate_ascendc\.sh"),
(_PY_INTERP, r"performance\.py"),
(_SH_INTERP, r"batch_run_performance\.sh"),
(_PY_INTERP, r"build_ascendc\.py"),
(_PY_INTERP, r"verification_ascendc\.py"),
(_PY_INTERP, r"verification_tilelang\.py"),
]

# 命令前缀允许的部分(环境变量赋值、cd ... &&、source ... &&)
# 这里只是用于推断"真实的脚本调用",不需要覆盖所有 shell 语法。
_LEADING_PREFIX = (
r"^\s*"
r"(?:export\s+)?" # 可选的 export 关键字
r"(?:[A-Za-z_][A-Za-z0-9_]*=\S+\s+)*" # ENV=VAL ...
r"(?:cd\s+\S+\s*&&\s*)?" # cd <dir> &&
r"(?:export\s+)?" # 可选的 export 关键字
r"(?:[A-Za-z_][A-Za-z0-9_]*=\S+\s+)*" # 再次允许 ENV=VAL
r"(?:&&\s+)*" # 支持 && 链式连接
)

# 项目根目录:优先从环境变量获取,fallback 到 __file__ 计算
# 避免 hook 被以相对路径调用时 __file__ 解析错误
PROJECT_ROOT = os.environ.get("PROJECT_ROOT")
if not PROJECT_ROOT:
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))


def should_intercept(command: str) -> bool:
# 仅当命令的"实际执行段"是 <interpreter> <path-to-script> 形式时才拦截。
# 兼容:环境变量前缀、可选的 `cd <dir> &&` 前缀。
for interp, script in INTERCEPTED_PATTERNS:
# 路径里允许出现 / 和非空白字符;脚本名用 \b 边界
pattern = (
_LEADING_PREFIX
+ interp
+ r"\s+"
+ r"(?:\S*/)?"
+ script
+ r"(?:\s|$)"
)
if re.search(pattern, command):
return True
return False


def emit_pretooluse(permission_decision: str, *, reason: str = "", additional_context: str = ""):
output = {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": permission_decision,
}
}
if reason:
output["hookSpecificOutput"]["permissionDecisionReason"] = reason
if additional_context:
output["hookSpecificOutput"]["additionalContext"] = additional_context
print(json.dumps(output, ensure_ascii=False), flush=True)


def truncate(text: str, limit: int = 8000) -> str:
if not text:
return ""
if len(text) <= limit:
return text
head = text[: limit // 2]
tail = text[-limit // 2 :]
return f"{head}\n\n... [truncated {len(text) - limit} chars] ...\n\n{tail}"


def execute_intercepted(command: str) -> None:
start_time = time.time()
cwd = os.getcwd()

# 解析路径:若是相对路径且在 cwd 不存在,则尝试 PROJECT_ROOT 下解析
# 这里我们直接通过 shell 执行原命令,但先确保 cwd 是 PROJECT_ROOT,
# 这样像 "python skills/..." 之类的相对路径能正确解析。
if os.path.isdir(PROJECT_ROOT):
cwd = PROJECT_ROOT

try:
proc = subprocess.run(
command,
shell=True,
cwd=cwd,
capture_output=True,
text=True,
timeout=1800,
)
exit_code = proc.returncode
stdout = proc.stdout or ""
stderr = proc.stderr or ""
except subprocess.TimeoutExpired as e:
exit_code = 124
stdout = (e.stdout.decode("utf-8", errors="replace") if e.stdout else "")
stderr = (e.stderr.decode("utf-8", errors="replace") if e.stderr else "")
stderr += "\n[HOOK] 命令执行超时(30 分钟)"
except Exception as e:
exit_code = 1
stdout = ""
stderr = f"[HOOK] 执行异常: {type(e).__name__}: {e}"

duration_ms = int((time.time() - start_time) * 1000)

additional_context = (
f"[skill_script_hook intercepted execution]\n"
f"command: {command}\n"
f"cwd: {cwd}\n"
f"exit_code: {exit_code}\n"
f"duration_ms: {duration_ms}\n"
f"--- stdout ---\n{truncate(stdout)}\n"
f"--- stderr ---\n{truncate(stderr)}\n"
)

status = "成功" if exit_code == 0 else "失败"
reason = (
f"[Hook 拦截提示] 该命令命中 skill_script_hook 拦截规则,已由 hook 代为执行({status},exit_code={exit_code})。"
f"执行结果见下方 additionalContext,原命令已被替换为 no-op,不会重复执行。"
)

# 使用 allow + updatedInput 将原命令替换为 no-op,避免 harness 显示 "Error:" 前缀
output = {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "allow",
"permissionDecisionReason": reason,
"additionalContext": additional_context,
"updatedInput": {
"command": f"echo '[hook-noop] original command was intercepted and executed by skill_script_hook'"
},
}
}
print(json.dumps(output, ensure_ascii=False), flush=True)


def read_stdin_command():
try:
raw = sys.stdin.read()
except Exception:
return None, None
if not raw or not raw.strip():
return None, None
try:
data = json.loads(raw)
except json.JSONDecodeError:
return None, None
if not isinstance(data, dict):
return None, None
tool_name = data.get("tool_name")
tool_input = data.get("tool_input") or {}
command = tool_input.get("command") if isinstance(tool_input, dict) else None
return tool_name, command


def main():
tool_name, command = read_stdin_command()

# 没有命令或不是 Bash —— allow,由 harness 处理
if not command or tool_name != "Bash":
emit_pretooluse("allow")
return

if should_intercept(command):
execute_intercepted(command)
else:
# 不拦截 —— 让 harness 正常处理(不在此重复执行)
emit_pretooluse("allow")


if __name__ == "__main__":
main()
16 changes: 16 additions & 0 deletions settings.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
{
"hooks": {
"PreToolUse": [
{
"matcher": "Bash",
"hooks": [
{
"type": "command",
"command": "PROJECT_ROOT=/home/ascendC python3 /home/ascendC/.claude/hooks/skill_script_hook.py",
"timeout": 1830
}
]
}
]
}
}
Loading