# 压缩操作手册

## 目标

把一个 Agent Skill 的工作型 Markdown 文档压短，但不破坏：
- 触发语义
- 输入输出契约
- 安全边界
- 默认路径与关键命令

## 推荐顺序

1. 先压 `SKILL.md`
2. 再压 `references/` 中真正会被频繁读取的文档
3. 默认不要把目标 skill 根目录下的 `README.md`、`CHANGELOG.md` 当成压缩目标

## 常用压缩手法

### 1. 删除重复解释

- 同一条规则只保留在最合适的一处
- `SKILL.md` 里保留摘要，细节放 `references/`
- 面向人类的说明文档默认不纳入压缩范围；若只为省字而改它们，往往是在压错对象

### 2. 缩短句子

- 一段只表达一个决策
- 把“原因 + 例外 + 重复铺垫”改成短句
- 长段落优先改为 3-5 条扁平 bullet

### 3. 合并低价值示例

- 保留最小可用示例
- 删除只是在换措辞、没有新增约束的信息
- 多个相近命令只保留主路径，其他写成“按需替换参数”

### 4. 主从分离

- `SKILL.md`：触发条件、主流程、硬约束、关键命令
- `references/`：详细检查清单、延伸案例、设计哲学

### 5. 把软建议和硬约束分开

- 硬约束必须留在主文档
- 经验性建议可以下沉到 `references/`

## 常见误伤

- 删除唯一的默认路径说明
- 把“不适用”删掉，导致 skill 误触发
- 把“必须执行”缩成“建议执行”
- 把输出文件名删掉，导致无法验收

## 简单判断标准

如果删掉一句话后，会让以下任一问题变得更难回答，就不要删：
- 什么时候触发这个 skill？
- 输入是什么？
- 输出是什么？
- 中间文件去哪？
- 哪些事不能做？

# 保留清单

压缩前后，至少逐项确认以下信息没有丢失。

## SKILL.md frontmatter

- `name`
- `description`
- `metadata.author`
- `metadata.keywords`
- `metadata.keywords` 仍包含 skill 名

## 触发与边界

- 什么时候应该用这个 skill
- 什么时候不该用
- 用户最小输入是什么
- 最终输出是什么

## 行为约束

- 默认工作区 / 隐藏目录
- 测试区或测试约束
- 只读/只写边界
- 必须执行的步骤
- 失败时的处理方式

## 关键实现信息

- 唯一或主要命令示例
- 关键脚本路径
- 配置文件路径或关键配置键
- 会影响行为的文件命名规则
- 默认待压缩范围是否仍限定在工作型 Markdown，而非 `README.md` / `CHANGELOG.md`

## 跨 skill 约定

- 与 `bensz-collect-bugs` 的协作约定
- 与 `parallel-vibe`、`auto-test-skill` 等强依赖 skill 的衔接方式

# 校验清单

## 结构校验

- `SKILL.md` 仍然存在
- frontmatter 合法
- 关键 Markdown 链接未失效，且没有跳出目标 skill 根目录
- `checked_files` 没有把目标 skill 根目录下的 `README.md`、`CHANGELOG.md` 当成默认压缩对象
- 中间文件都在 `.compact-bensz-skills/run-{timestamp}/`
- `latest-run.txt` 指向的是本次验证对应的 run

## 语义校验

- 触发条件没有变窄或变偏
- 不适用范围仍然存在
- 安全限制仍然清楚
- 输入/输出/默认目录仍然可回答

## 收益校验

- 总字数下降
- 最大文件的字数明显下降，或有合理解释
- `SKILL.md` 比修改前更短、更聚焦

## 风险提示

如果出现以下情况，应停止并回看快照：
- frontmatter 被改坏
- 关键命令消失
- 输出文件名或路径说明消失
- 只读/只写限制被删
- 关键链接改成了越界相对路径

from __future__ import annotations

import fnmatch
import json
import re
import shutil
from datetime import datetime
from pathlib import Path
from typing import Any

try:
    import yaml
except ModuleNotFoundError as exc:  # pragma: no cover - import guard
    raise SystemExit(
        "缺少 PyYAML 依赖，无法读取 compact-bensz-skills/config.yaml。请先安装 pyyaml。"
    ) from exc


FRONTMATTER_RE = re.compile(r"\A---\n(.*?)\n---\n?", re.DOTALL)
WORD_RE = re.compile(r"[A-Za-z0-9_]+|[\u4e00-\u9fff]")
HEADING_RE = re.compile(r"^#{1,6}\s", re.MULTILINE)
FENCE_RE = re.compile(r"^```", re.MULTILINE)
LOCAL_LINK_RE = re.compile(r"\[[^\]]+\]\(([^)]+)\)")


def skill_dir() -> Path:
    return Path(__file__).resolve().parent.parent


def load_config() -> dict[str, Any]:
    config_path = skill_dir() / "config.yaml"
    return yaml.safe_load(config_path.read_text(encoding="utf-8"))


def resolve_skill_root(skill_root: str) -> Path:
    path = Path(skill_root).expanduser().resolve()
    if not path.is_dir():
        raise SystemExit(f"skill_root 不存在或不是目录: {path}")
    if not (path / "SKILL.md").exists():
        raise SystemExit(f"目标目录缺少 SKILL.md: {path}")
    return path


def resolve_workspace_root(
    target_skill_root: Path,
    config: dict[str, Any],
    workspace_dir: str | None = None,
    run_id: str | None = None,
    create: bool = False,
) -> Path:
    workspace_base, explicit_run_id = resolve_workspace_base(target_skill_root, config, workspace_dir)
    effective_run_id = run_id or explicit_run_id
    if effective_run_id is None:
        if create:
            effective_run_id = allocate_unique_run_id(workspace_base, config, generate_run_id(config))
        else:
            effective_run_id = read_latest_run_id(workspace_base, config)
            if effective_run_id is None:
                raise SystemExit(
                    "未找到可复用的 run 目录。请先运行 init_workspace.py，或显式传入 --run-id。"
                )
    return workspace_base / effective_run_id


def resolve_workspace_base(
    target_skill_root: Path,
    config: dict[str, Any],
    workspace_dir: str | None = None,
) -> tuple[Path, str | None]:
    if workspace_dir:
        candidate = Path(workspace_dir).expanduser().resolve()
        if is_run_dir_name(candidate.name, config):
            return candidate.parent, candidate.name
        return candidate, None
    return target_skill_root / config["workspace"]["hidden_dir"], None


def generate_run_id(config: dict[str, Any]) -> str:
    prefix = config["workspace"]["run_prefix"]
    timestamp_format = config["workspace"]["timestamp_format"]
    return f"{prefix}{datetime.now().strftime(timestamp_format)}"


def is_run_dir_name(name: str, config: dict[str, Any]) -> bool:
    prefix = config["workspace"]["run_prefix"]
    timestamp_format = config["workspace"]["timestamp_format"]
    sample = datetime(2000, 1, 2, 3, 4, 5).strftime(timestamp_format)
    digit_pattern = "".join(r"\d" if ch.isdigit() else re.escape(ch) for ch in sample)
    pattern = rf"^{re.escape(prefix)}{digit_pattern}(?:-\d{{2}})?$"
    return re.match(pattern, name) is not None


def allocate_unique_run_id(workspace_base: Path, config: dict[str, Any], base_run_id: str) -> str:
    if not (workspace_base / base_run_id).exists():
        return base_run_id
    for idx in range(2, 100):
        candidate = f"{base_run_id}-{idx:02d}"
        if not (workspace_base / candidate).exists():
            return candidate
    raise SystemExit(f"无法在 {workspace_base} 下分配唯一 run 目录: {base_run_id}")


def latest_run_pointer_path(workspace_base: Path, config: dict[str, Any]) -> Path:
    return workspace_base / config["workspace"]["latest_run_pointer"]


def write_latest_run_id(workspace_base: Path, config: dict[str, Any], run_id: str) -> None:
    pointer = latest_run_pointer_path(workspace_base, config)
    pointer.parent.mkdir(parents=True, exist_ok=True)
    pointer.write_text(run_id + "\n", encoding="utf-8")


def read_latest_run_id(workspace_base: Path, config: dict[str, Any]) -> str | None:
    pointer = latest_run_pointer_path(workspace_base, config)
    if not pointer.exists():
        return None
    run_id = pointer.read_text(encoding="utf-8").strip()
    return run_id or None


def path_within(base_dir: Path, target: Path) -> bool:
    try:
        target.resolve().relative_to(base_dir.resolve())
        return True
    except ValueError:
        return False


def workspace_subpaths(config: dict[str, Any], workspace_root: Path) -> dict[str, Path]:
    subpaths = {name: workspace_root / name for name in config["workspace"]["subdirs"]}
    subpaths["workspace_root"] = workspace_root
    subpaths["workspace_base"] = workspace_root.parent
    subpaths["before_snapshot_dir"] = workspace_root / config["workspace"]["before_snapshot_dir"]
    for key, relative_path in config["reports"].items():
        subpaths[key] = workspace_root / relative_path
    return subpaths


def ensure_workspace(config: dict[str, Any], workspace_root: Path) -> dict[str, Path]:
    paths = workspace_subpaths(config, workspace_root)
    for path in paths.values():
        if path == paths["workspace_base"]:
            path.mkdir(parents=True, exist_ok=True)
            continue
        if path.suffix:
            path.parent.mkdir(parents=True, exist_ok=True)
        else:
            path.mkdir(parents=True, exist_ok=True)
    return paths


def should_ignore_path(relative_path: Path, config: dict[str, Any]) -> bool:
    ignore_dirs = set(config["source_files"]["ignore_dirs"])
    if any(part in ignore_dirs for part in relative_path.parts[:-1]):
        return True
    ignore_file_names = set(config["source_files"].get("ignore_file_names", []))
    if relative_path.name in ignore_file_names:
        return True
    ignore_globs = config["source_files"]["ignore_globs"]
    return any(fnmatch.fnmatch(relative_path.name, pattern) for pattern in ignore_globs)


def iter_markdown_files(
    target_skill_root: Path,
    config: dict[str, Any],
    workspace_root: Path,
) -> list[Path]:
    markdown_exts = set(config["source_files"]["markdown_extensions"])
    files: list[Path] = []
    for path in sorted(target_skill_root.rglob("*")):
        if not path.is_file():
            continue
        if path.suffix.lower() not in markdown_exts:
            continue
        try:
            relative_path = path.relative_to(target_skill_root)
        except ValueError:
            continue
        if workspace_root in path.parents or path == workspace_root:
            continue
        if should_ignore_path(relative_path, config):
            continue
        files.append(path)
    return files


def read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def compute_text_stats(text: str) -> dict[str, int]:
    return {
        "chars": len(text),
        "words": len(WORD_RE.findall(text)),
        "lines": text.count("\n") + (0 if not text else 1),
        "headings": len(HEADING_RE.findall(text)),
        "fence_markers": len(FENCE_RE.findall(text)),
    }


def compute_file_record(target_skill_root: Path, path: Path) -> dict[str, Any]:
    text = read_text(path)
    record = {"path": str(path.relative_to(target_skill_root))}
    record.update(compute_text_stats(text))
    return record


def build_totals(records: list[dict[str, Any]], phase: str) -> dict[str, Any]:
    return {
        "phase": phase,
        "file_count": len(records),
        "total_chars": sum(int(record["chars"]) for record in records),
        "total_words": sum(int(record["words"]) for record in records),
        "total_lines": sum(int(record["lines"]) for record in records),
        "total_headings": sum(int(record["headings"]) for record in records),
        "total_fence_markers": sum(int(record["fence_markers"]) for record in records),
        "files": records,
    }


def write_json(path: Path, payload: dict[str, Any] | list[Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=False) + "\n",
        encoding="utf-8",
    )


def write_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")


def snapshot_files(target_skill_root: Path, files: list[Path], snapshot_root: Path) -> None:
    if snapshot_root.exists():
        shutil.rmtree(snapshot_root)
    for file_path in files:
        relative_path = file_path.relative_to(target_skill_root)
        destination = snapshot_root / relative_path
        destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(file_path, destination)


def parse_frontmatter(text: str) -> tuple[dict[str, Any], str]:
    match = FRONTMATTER_RE.match(text)
    if not match:
        return {}, text
    data = yaml.safe_load(match.group(1)) or {}
    body = text[match.end() :]
    return data, body


def nested_get(data: dict[str, Any], dotted_key: str) -> Any:
    value: Any = data
    for part in dotted_key.split("."):
        if not isinstance(value, dict) or part not in value:
            return None
        value = value[part]
    return value


def normalize_link_target(target: str) -> str | None:
    cleaned = target.strip()
    if not cleaned or cleaned.startswith("#"):
        return None
    if "://" in cleaned or cleaned.startswith("mailto:"):
        return None
    return cleaned.split("#", 1)[0]


def find_local_link_issues(
    base_file: Path,
    text: str,
    target_skill_root: Path,
) -> list[str]:
    issues: list[str] = []
    for raw_target in LOCAL_LINK_RE.findall(text):
        normalized = normalize_link_target(raw_target)
        if not normalized:
            continue
        resolved = (base_file.parent / normalized).resolve()
        if not path_within(target_skill_root, resolved):
            issues.append(f"本地链接越出 skill 根目录 -> {raw_target}")
            continue
        if not resolved.exists():
            issues.append(f"本地链接不存在 -> {raw_target}")
    return issues

#!/usr/bin/env python3
from __future__ import annotations

import argparse

from common import (
    build_totals,
    compute_file_record,
    ensure_workspace,
    iter_markdown_files,
    load_config,
    path_within,
    resolve_skill_root,
    resolve_workspace_root,
    snapshot_files,
    write_latest_run_id,
    write_json,
    write_text,
)


def build_plan(records: list[dict[str, object]]) -> str:
    top_records = sorted(records, key=lambda item: item["words"], reverse=True)[:5]
    lines = [
        "# 压缩计划",
        "",
        "## 当前判断",
        "",
        "- 先阅读 `SKILL.md`、`config.yaml` 与关键脚本，再决定哪些文档可以缩短。",
        "- `tests/`、`plans/`、`README.md`、`CHANGELOG.md` 已被排除，不作为默认待压缩源文件。",
        "- 优先压缩字数最高、重复解释最多的 Markdown 文件。",
        "",
        "## 优先处理文件",
        "",
    ]
    if not top_records:
        lines.append("- 暂无待处理 Markdown 文件")
    else:
        for record in top_records:
            lines.append(f"- `{record['path']}`：约 {record['words']} 词，{record['chars']} 字符")
    lines.extend(
        [
            "",
            "## 必保留信息",
            "",
            "- frontmatter 中的 `name`、`description`、`metadata.author`",
            "- 关键命令、关键路径、默认输出",
            "- 安全限制与不适用范围",
            "",
            "## 完成后必做",
            "",
            "- 重新运行 `measure_markdown.py --phase after`",
            "- 运行 `validate_compaction.py`",
        ]
    )
    return "\n".join(lines) + "\n"


def main() -> None:
    parser = argparse.ArgumentParser(description="初始化 compact-bensz-skills 隐藏工作区")
    parser.add_argument("--skill-root", required=True, help="目标 skill 根目录")
    parser.add_argument("--workspace-dir", help="自定义隐藏工作区")
    parser.add_argument("--run-id", help="显式指定 run 目录名，例如 run-20260328194500")
    args = parser.parse_args()

    config = load_config()
    target_skill_root = resolve_skill_root(args.skill_root)
    workspace_root = resolve_workspace_root(
        target_skill_root,
        config,
        args.workspace_dir,
        run_id=args.run_id,
        create=True,
    )
    paths = ensure_workspace(config, workspace_root)
    write_latest_run_id(paths["workspace_base"], config, workspace_root.name)

    markdown_files = iter_markdown_files(target_skill_root, config, workspace_root)
    snapshot_files(target_skill_root, markdown_files, paths["before_snapshot_dir"])
    records = [compute_file_record(target_skill_root, path) for path in markdown_files]
    inventory = {
        "skill_root": str(target_skill_root),
        "workspace_base": str(paths["workspace_base"]),
        "workspace_root": str(workspace_root),
        "run_id": workspace_root.name,
        "workspace_inside_skill_root": path_within(target_skill_root, workspace_root),
        "file_count": len(records),
        "files": records,
    }
    totals = build_totals(records, phase="before")

    write_json(paths["inventory_json"], inventory)
    write_json(paths["size_before_json"], totals)
    write_text(paths["plan_markdown"], build_plan(records))

    print(f"workspace_base={paths['workspace_base']}")
    print(f"workspace_root={workspace_root}")
    print(f"run_id={workspace_root.name}")
    print(f"inventory={paths['inventory_json']}")
    print(f"snapshot_dir={paths['before_snapshot_dir']}")
    print(f"size_before={paths['size_before_json']}")
    if not inventory["workspace_inside_skill_root"]:
        print("warning=workspace_dir 位于目标 skill 根目录之外；这只应在用户明确指定时使用")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

import argparse

from common import (
    build_totals,
    compute_file_record,
    ensure_workspace,
    iter_markdown_files,
    load_config,
    resolve_skill_root,
    resolve_workspace_root,
    write_json,
    write_text,
)


def format_delta(before: dict[str, int], after: dict[str, int]) -> str:
    delta_words = after["total_words"] - before["total_words"]
    delta_chars = after["total_chars"] - before["total_chars"]
    lines = [
        "# 压缩统计对比",
        "",
        f"- 压缩前总词数：{before['total_words']}",
        f"- 压缩后总词数：{after['total_words']}",
        f"- 词数变化：{delta_words}",
        f"- 压缩前总字符数：{before['total_chars']}",
        f"- 压缩后总字符数：{after['total_chars']}",
        f"- 字符变化：{delta_chars}",
        "",
        "## 当前最大文件",
        "",
    ]
    for record in sorted(after["files"], key=lambda item: item["words"], reverse=True)[:5]:
        lines.append(f"- `{record['path']}`：{record['words']} 词，{record['chars']} 字符")
    return "\n".join(lines) + "\n"


def main() -> None:
    parser = argparse.ArgumentParser(description="统计目标 skill Markdown 体积")
    parser.add_argument("--skill-root", required=True, help="目标 skill 根目录")
    parser.add_argument(
        "--phase",
        default="after",
        choices=["before", "after"],
        help="写入前/后统计文件",
    )
    parser.add_argument("--workspace-dir", help="自定义隐藏工作区")
    parser.add_argument("--run-id", help="复用 init_workspace.py 创建的 run 目录名")
    args = parser.parse_args()

    config = load_config()
    target_skill_root = resolve_skill_root(args.skill_root)
    workspace_root = resolve_workspace_root(
        target_skill_root,
        config,
        args.workspace_dir,
        run_id=args.run_id,
    )
    paths = ensure_workspace(config, workspace_root)
    markdown_files = iter_markdown_files(target_skill_root, config, workspace_root)
    records = [compute_file_record(target_skill_root, path) for path in markdown_files]
    totals = build_totals(records, phase=args.phase)

    output_key = "size_before_json" if args.phase == "before" else "size_after_json"
    write_json(paths[output_key], totals)

    before_path = paths["size_before_json"]
    after_path = paths["size_after_json"]
    if before_path.exists() and after_path.exists():
        import json

        before = json.loads(before_path.read_text(encoding="utf-8"))
        after = json.loads(after_path.read_text(encoding="utf-8"))
        write_text(paths["size_delta_markdown"], format_delta(before, after))

    print(f"workspace_root={workspace_root}")
    print(f"run_id={workspace_root.name}")
    print(f"{args.phase}_stats={paths[output_key]}")
    if paths["size_delta_markdown"].exists():
        print(f"delta_report={paths['size_delta_markdown']}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
from pathlib import Path

from common import (
    build_totals,
    compute_file_record,
    ensure_workspace,
    find_local_link_issues,
    iter_markdown_files,
    load_config,
    nested_get,
    path_within,
    parse_frontmatter,
    read_text,
    resolve_skill_root,
    resolve_workspace_root,
    write_json,
)


def main() -> None:
    parser = argparse.ArgumentParser(description="校验压缩后的 skill Markdown")
    parser.add_argument("--skill-root", required=True, help="目标 skill 根目录")
    parser.add_argument("--workspace-dir", help="自定义隐藏工作区")
    parser.add_argument("--run-id", help="复用 init_workspace.py 创建的 run 目录名")
    args = parser.parse_args()

    config = load_config()
    target_skill_root = resolve_skill_root(args.skill_root)
    workspace_root = resolve_workspace_root(
        target_skill_root,
        config,
        args.workspace_dir,
        run_id=args.run_id,
    )
    paths = ensure_workspace(config, workspace_root)
    markdown_files = iter_markdown_files(target_skill_root, config, workspace_root)

    errors: list[str] = []
    warnings: list[str] = []
    skill_md = target_skill_root / "SKILL.md"
    frontmatter, skill_body = parse_frontmatter(read_text(skill_md))

    for dotted_key in config["preservation"]["required_skill_frontmatter_fields"]:
        if nested_get(frontmatter, dotted_key) in (None, "", []):
            errors.append(f"SKILL.md frontmatter 缺少必需字段: {dotted_key}")

    if nested_get(frontmatter, "name") != target_skill_root.name:
        warnings.append(
            "SKILL.md frontmatter.name 与目录名不一致；若这是历史兼容设计可忽略，否则建议修正"
        )

    keywords = nested_get(frontmatter, "metadata.keywords") or []
    if config["preservation"]["required_skill_keywords_include_name"] and nested_get(
        frontmatter, "name"
    ) not in keywords:
        errors.append("metadata.keywords 未包含 skill 名")

    if not path_within(target_skill_root, workspace_root):
        warnings.append("workspace_dir 位于目标 skill 根目录之外；只有用户明确指定时才应这样做")

    for path in markdown_files:
        issues = find_local_link_issues(path, read_text(path), target_skill_root)
        errors.extend(
            [f"{path.relative_to(target_skill_root)}: {issue}" for issue in issues]
        )

    if not skill_body.strip():
        errors.append("SKILL.md 正文为空")

    before_stats_path = paths["size_before_json"]
    after_stats_path = paths["size_after_json"]
    if before_stats_path.exists():
        before = json.loads(before_stats_path.read_text(encoding="utf-8"))
        if not after_stats_path.exists():
            current_records = [compute_file_record(target_skill_root, path) for path in markdown_files]
            write_json(after_stats_path, build_totals(current_records, phase="after"))
        after = json.loads(after_stats_path.read_text(encoding="utf-8"))
        if after["total_words"] >= before["total_words"]:
            warnings.append(
                "压缩后总词数未下降；如果这是有意保留，请在报告中说明原因"
            )

    snapshot_skill_md = paths["before_snapshot_dir"] / "SKILL.md"
    if snapshot_skill_md.exists():
        snapshot_frontmatter, _ = parse_frontmatter(read_text(snapshot_skill_md))
        if nested_get(snapshot_frontmatter, "name") != nested_get(frontmatter, "name"):
            errors.append("SKILL.md frontmatter.name 与压缩前快照不一致")
        if nested_get(snapshot_frontmatter, "description") != nested_get(frontmatter, "description"):
            warnings.append("SKILL.md frontmatter.description 与压缩前快照不同；请确认只是等价压缩")

    result = {
        "skill_root": str(target_skill_root),
        "workspace_base": str(paths["workspace_base"]),
        "workspace_root": str(workspace_root),
        "run_id": workspace_root.name,
        "workspace_inside_skill_root": path_within(target_skill_root, workspace_root),
        "checked_files": [str(path.relative_to(target_skill_root)) for path in markdown_files],
        "error_count": len(errors),
        "warning_count": len(warnings),
        "errors": errors,
        "warnings": warnings,
    }
    write_json(paths["validation_json"], result)

    print(f"workspace_root={workspace_root}")
    print(f"run_id={workspace_root.name}")
    print(f"validation={paths['validation_json']}")
    if warnings:
        print(f"warnings={len(warnings)}")
    if errors:
        for error in errors:
            print(f"ERROR: {error}")
        raise SystemExit(1)


if __name__ == "__main__":
    main()