Harness² + CodeX 多代码库实践指南：构建企业级 AI 编程Harness与自进化体系

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

D:\Company\Codes\LCM\                   # 多代码库总根目录（工作区）
├── AGENTS.md                          # [核心指令与总Harness] 全局规范、多仓协同规则、技能自动触发索引 (<8KB)
├── .agents/
│   └── skills/                        # [按需动态技能库] 只有触发匹配时才加载对应内容
│       ├── tdd-workflow/
│       │   └── SKILL.md               # 技能1：Spring Boot + Django 测试先行与回归
│       ├── code-review/
│       │   └── SKILL.md               # 技能2：高标准代码审查（事务、空安全、N+1）
│       ├── diagnosing-bugs/
│       │   └── SKILL.md               # 技能3：跨服务调用、Bean冲突、Celery故障排查
│       ├── multi-repo-sync/
│       │   └── SKILL.md               # 技能4：跨模块依赖查找与 Python/Java REST DTO 兼容
│       └── vm-e2e/
│           └── SKILL.md               # 技能5：端到端集成测试、状态账本验证与执行入口
├── scripts/                           # [确定性辅助工具] 消除 AI 随意拼凑命令的不确定性
│   ├── check_workspace.py             # 脚本1：多代码库分支与未提交变动一键巡检
│   ├── run_checks.py                  # 脚本2：跨仓库编译/类型检查/单元测试/E2E统一分级执行器
│   └── reflect_experience.py          # 脚本3：Harness² 经验反思器（自动提炼经验写回 AGENTS.md）
├── sub_agents/                        # [独立角色代理]
│   ├── architect/
│   │   └── AGENT.md                   # 架构师角色：负责跨库接口设计与重构方案制定
│   └── reviewer/
│       └── AGENT.md                   # 审查员角色：负责质量验收与提交规范审查
├── .codex/
│   └── hooks.json                     # [生命周期钩子] 拦截危险命令 (rm -rf, git reset) + 语法格式化
├── python-connect/                    # 子仓库 A：Python 业务服务
├── platform-lcm/                      # 子仓库 B：Java Spring Boot 核心业务
├── sdwan-manager/                     # 子仓库 C：Java 集群控制服务
└── platform-support-design/           # 子仓库 D：底层 VM 生命周期端到端测试仓

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# Multi-Repo Master Engineering Harness (Harness² Standard)

## 1. Workspace Topology & Repositories
- `python-connect`: Python/Django/Celery service.
- `platform-lcm`: Java Spring Boot core management service.
- `sdwan-manager`: Java cluster controller.
- `platform-support-design`: End-to-end VM lifecycle test suite and tools.

## 2. Universal Engineering Rules
- **No Direct Destructive Commands**: Never execute `git reset --hard`, `git push --force`, or unverified deletions.
- **Verification First**: Always run local verification via `scripts/run_checks.py` before completing any task.
- **Multi-Repo DTO Consistency**: When modifying API payloads, maintain backward compatibility across Java and Python.

## 3. Dynamic Skills Trigger Mapping
Read the corresponding SKILL.md ONLY when the task matches the trigger:
- When writing new features or fixing bugs, read `.agents/skills/tdd-workflow/SKILL.md`.
- When reviewing pull requests or code quality, read `.agents/skills/code-review/SKILL.md`.
- When diagnosing complex exceptions or stack traces, read `.agents/skills/diagnosing-bugs/SKILL.md`.
- When modifying cross-repo APIs or DTOs between Python and Java, read `.agents/skills/multi-repo-sync/SKILL.md`.
- When preparing or running VM lifecycle End-to-End integration tests, read `.agents/skills/vm-e2e/SKILL.md`.

## 4. Deterministic Scripts Contract
- Workspace check: `python scripts/check_workspace.py`
- Fast checks: `python scripts/run_checks.py [all|python|ecr|v2-mgr|e2e]`
- Reflection & Lesson Record: `python scripts/reflect_experience.py --category <Category> --lesson "<Lesson>"`

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

---
name: multi-repo-sync
description: Cross-repository API, REST DTO, and data contract alignment between Python and Java
---

# Multi-Repo Contract Synchronization Workflow

## 1. Inspection Checklist
1. Identify the Producer and Consumer:
   - If Python produces JSON: Check Jackson deserialization annotations (`@JsonProperty`) in Java.
   - If Java produces JSON: Check serializer/pydantic models in Python.
2. Field naming conventions:
   - Ensure case mappings (`camelCase` vs `snake_case`) are explicitly declared.
3. Enum sync:
   - When adding a state in Java enum, immediately sync the Python choice enum.

## 2. Verification Protocol
1. Modify the DTO definitions in both repositories simultaneously.
2. Run unit tests in both repositories using `python scripts/run_checks.py python` and `python scripts/run_checks.py ecr`.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

---
name: vm-e2e
description: End-to-End VM lifecycle integration testing via platform-support-design tools
---

# VM Lifecycle E2E Testing Protocol

## Core Rules
1. **Never use test backdoors**: All tests must invoke the actual public APIs.
2. **HTTP 200 != Success**: Always verify against the state ledger (database & NetBox records) to confirm real VM provisioning.
3. **Platform Token Lifecycle**: Long-term reuse of existing environment tokens; do not generate transient tokens.

## Execution Procedure
- Preflight environment check:
  `python platform-support-design/tools/e2e.py check --env <profile.json>`
- Run Smoke test suite:
  `python platform-support-design/tools/e2e.py run --env <profile.json> --suite smoke`
- Cleanup test resources safely:
  `python platform-support-design/tools/e2e.py clean --env <profile.json>`

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

#!/usr/bin/env python3
"""
scripts/check_workspace.py
Inspect git branches, working tree states, and recent commits across all sub-repos.
"""
import subprocess
from pathlib import Path

WORKSPACE_ROOT = Path(__file__).resolve().parent.parent

def check_repo(repo_dir: Path):
    if not (repo_dir / ".git").exists():
        return None
    try:
        branch = subprocess.check_output(
            ["git", "branch", "--show-current"], cwd=repo_dir, text=True
        ).strip() or "DETACHED"
        status_out = subprocess.check_output(
            ["git", "status", "--porcelain"], cwd=repo_dir, text=True
        )
        modified = staged = untracked = 0
        for line in status_out.splitlines():
            if not line:
                continue
            x, y = line[0], line[1]
            if x in "MADRC":
                staged += 1
            if y in "MADRC":
                modified += 1
            if x == "?" and y == "?":
                untracked += 1
        commit = subprocess.check_output(
            ["git", "log", "-1", "--format=%h %s"], cwd=repo_dir, text=True
        ).strip()[:30]
        return {
            "name": repo_dir.name,
            "branch": branch,
            "msu": f"{modified}/{staged}/{untracked}",
            "commit": commit,
        }
    except Exception as e:
        return {"name": repo_dir.name, "branch": "ERR", "msu": "-", "commit": str(e)}

def main():
    print("=" * 80)
    print(f"{'Repository':<25} | {'Branch':<20} | {'M/S/U':<8} | {'Recent Commit'}")
    print("-" * 80)
    for item in sorted(WORKSPACE_ROOT.iterdir()):
        if item.is_dir() and (item / ".git").exists():
            info = check_repo(item)
            if info:
                print(f"{info['name']:<25} | {info['branch']:<20} | {info['msu']:<8} | {info['commit']}")
    print("=" * 80)
    print("M/S/U: Modified / Staged / Untracked files")

if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

#!/usr/bin/env python3
"""
scripts/run_checks.py
Unified tiered verification runner for Java, Python, and E2E testing suites.
"""
import sys
import subprocess
from pathlib import Path

WORKSPACE = Path(__file__).resolve().parent.parent

def run_cmd(cmd, cwd):
    print(f"\n>>> Running: {' '.join(cmd)} in {cwd}")
    ret = subprocess.run(cmd, cwd=cwd)
    return ret.returncode == 0

def check_python():
    py_dir = WORKSPACE / "python-connect"
    return run_cmd(["pytest", "-m", "unit"], cwd=py_dir)

def check_ecr():
    ecr_dir = WORKSPACE / "platform-lcm"
    mvn_cmd = "mvn.cmd" if sys.platform == "win32" else "mvn"
    return run_cmd([mvn_cmd, "test-compile"], cwd=ecr_dir)

def check_e2e():
    e2e_runner = WORKSPACE / "platform-support-design" / "tools" / "e2e.py"
    print(f"\n>>> Preflight E2E check using runner: {e2e_runner}")
    return run_cmd([sys.executable, str(e2e_runner), "--help"], cwd=WORKSPACE)

def main():
    target = sys.argv[1] if len(sys.argv) > 1 else "all"
    results = {}
    if target in ("python", "all"):
        results["python"] = check_python()
    if target in ("ecr", "all"):
        results["ecr"] = check_ecr()
    if target in ("e2e", "all"):
        results["e2e"] = check_e2e()

    print("\n" + "=" * 40)
    print("Verification Summary:")
    for k, ok in results.items():
        print(f"  {k:<14}: {'PASS' if ok else 'FAIL'}")
    print("=" * 40)
    if not all(results.values()):
        sys.exit(1)

if __name__ == "__main__":
    main()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

{
  "version": "1.0",
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "tool_name == 'execute_bash' && (args.command.contains('reset --hard') || args.command.contains('--force'))",
        "action": "BLOCK",
        "message": "Security Block: Destructive Git commands (reset --hard, force push) are strictly prohibited."
      },
      {
        "matcher": "tool_name == 'execute_bash' && args.command.contains('rm -rf /')",
        "action": "BLOCK",
        "message": "Security Block: High-risk recursive deletion intercepted."
      }
    ],
    "PostToolUse": [
      {
        "matcher": "tool_name == 'edit_file' && (file_path.endsWith('.java') || file_path.endsWith('.py'))",
        "action": "NOTIFY",
        "message": "Source code edited. Please ensure formatting passes and run 'python scripts/run_checks.py' for quick regression check."
      }
    ]
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

#!/usr/bin/env python3
"""
scripts/reflect_experience.py
Append lessons learned during debugging directly into AGENTS.md.
"""
import argparse
from datetime import datetime
from pathlib import Path

AGENTS_FILE = Path(__file__).resolve().parent.parent / "AGENTS.md"

def append_lesson(category: str, lesson: str):
    if not AGENTS_FILE.exists():
        print(f"Error: {AGENTS_FILE} does not exist.")
        return

    content = AGENTS_FILE.read_text(encoding="utf-8")
    header = "## 5. Lessons Learned & Troubleshooting Ledger (Auto-Reflected)"
    
    timestamp = datetime.now().strftime("%Y-%m-%d")
    record = f"- [{timestamp}] **[{category}]**: {lesson}\n"

    if header not in content:
        content += f"\n\n{header}\n{record}"
    else:
        parts = content.split(header)
        content = parts[0] + header + "\n" + record + parts[1].lstrip("\n")

    AGENTS_FILE.write_text(content, encoding="utf-8")
    print(f"Successfully recorded lesson into AGENTS.md under [{category}].")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Harness² Reflection Recorder")
    parser.add_argument("--category", required=True, help="Category (e.g., Java, Python, E2E, Git)")
    parser.add_argument("--lesson", required=True, help="Detailed lesson or pitfall to avoid")
    args = parser.parse_args()

    append_lesson(args.category, args.lesson)

1

python scripts/reflect_experience.py --category Python --lesson "Celery 5.x 传输 LocalDateTime 时必须显式注册 ISO8601 json 编码器，否则报 Object of type datetime is not JSON serializable"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

[用户任务输入]
     │
     ▼
[Harness Editor (Proposer)]  ──自动针对该任务生成 2~4 个候选提示词/测试脚本
     │
     ▼
[Parallel Probing]           ──并行启动 4 个独立的 Codex 进程分别试跑该任务
     │
     ▼
[Harness Composer]           ──比较 4 个执行轨迹，合成出最终的最优代码