Compare commits

...

9 Commits

Author SHA1 Message Date
swe-agent[bot]
b227fee225 fix codeagent claude and gemini root dir 2025-12-13 16:56:53 +08:00
swe-agent[bot]
2b7569335b update readme 2025-12-13 15:29:12 +08:00
swe-agent[bot]
9e667f0895 feat(v5.2.0): Complete skills system integration and config cleanup
Core Changes:
- **Skills System**: Added codeagent, product-requirements, prototype-prompt-generator to skill-rules.json
- **Config Cleanup**: Removed deprecated gh module from config.json
- **Workflow Update**: Changed memorys/CLAUDE.md to use codeagent skill instead of codex

Details:
- config.json:88-119: Removed gh module (github-workflow directory doesn't exist)
- skills/skill-rules.json:24-114: Added 3 new skills with keyword/pattern triggers
  - codeagent: multi-backend, parallel task execution
  - product-requirements: PRD, requirements gathering
  - prototype-prompt-generator: UI/UX design specifications
- memorys/CLAUDE.md:3,24-25: Updated Codex skill → Codeagent skill

Verification:
- All skill activation tests PASS
- codeagent skill triggers correctly on keyword match

Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-13 13:25:21 +08:00
swe-agent[bot]
4759eb2c42 chore(v5.2.0): Update CHANGELOG and remove deprecated test files
- Added Skills System Enhancements section to CHANGELOG
- Documented new skills: codeagent, product-requirements, prototype-prompt-generator
- Removed deprecated test files (tests/test_*.py)
- Updated release date to 2025-12-13

Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-13 13:21:59 +08:00
swe-agent[bot]
edbf168b57 fix(codeagent-wrapper): fix race condition in stdout parsing
修复 GitHub Actions CI 中的测试失败问题。

问题分析:
在 TestRun_PipedTaskSuccess 测试中,当脚本运行很快时,cmd.Wait()
可能在 parseJSONStreamInternal goroutine 开始读取之前就返回,
导致 stdout 管道被过早关闭,出现 "read |0: file already closed" 错误。

解决方案:
将 parseJSONStreamInternal goroutine 的启动提前到 cmd.Start() 之前。
这确保解析器在进程启动前就 ready,避免竞态条件。

测试结果:
- 本地所有测试通过 ✓
- 覆盖率保持 93.7% ✓

Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-13 13:20:49 +08:00
swe-agent[bot]
9bfea81ca6 docs(changelog): remove GitHub workflow related content
GitHub workflow features have been removed from the project.

Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-13 13:01:06 +08:00
swe-agent[bot]
a9bcea45f5 Merge rc/5.2 into master: v5.2.0 release improvements 2025-12-13 12:56:37 +08:00
swe-agent[bot]
8554da6e2f feat(v5.2.0): Improve release notes and installation scripts
**Main Changes**:
1. release.yml: Extract version release notes from CHANGELOG.md
2. install.bat: codex-wrapper → codeagent-wrapper
3. README.md: Update multi-backend architecture description
4. README_CN.md: Sync Chinese description
5. CHANGELOG.md: Complete v5.2.0 release notes in English

Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-13 12:53:28 +08:00
ben
b2f941af5f Merge pull request #53 from cexll/rc/5.2
feat: Enterprise Workflow with Multi-Backend Support (v5.2)
2025-12-13 12:38:38 +08:00
17 changed files with 293 additions and 922 deletions

View File

@@ -104,10 +104,38 @@ jobs:
cp install.sh install.bat release/
ls -la release/
- name: Extract release notes from CHANGELOG
id: extract_notes
run: |
VERSION=${GITHUB_REF#refs/tags/v}
# Extract version section from CHANGELOG.md
awk -v ver="$VERSION" '
/^## [0-9]+\.[0-9]+\.[0-9]+ - / {
if (found) exit
if ($2 == ver) {
found = 1
next
}
}
found && /^## / { exit }
found { print }
' CHANGELOG.md > release_notes.md
# Fallback to auto-generated if extraction failed
if [ ! -s release_notes.md ]; then
echo "⚠️ No release notes found in CHANGELOG.md for version $VERSION" > release_notes.md
echo "" >> release_notes.md
echo "## What's Changed" >> release_notes.md
echo "See commits in this release for details." >> release_notes.md
fi
cat release_notes.md
- name: Create Release
uses: softprops/action-gh-release@v2
with:
files: release/*
generate_release_notes: true
body_path: release_notes.md
draft: false
prerelease: false

View File

@@ -1,6 +1,129 @@
# Changelog
## 5.2.0 - 2025-12-11
- PR #53: unified CLI version source in `codeagent-wrapper/main.go` and bumped to 5.2.0.
- Added legacy `codex-wrapper` alias support (runtime detection plus `scripts/install.sh` symlink helper).
- Updated documentation to reflect backend flag usage and new version output.
## 5.2.0 - 2025-12-13
### 🚀 Core Features
#### Skills System Enhancements
- **New Skills**: Added `codeagent`, `product-requirements`, `prototype-prompt-generator` to `skill-rules.json`
- **Auto-Activation**: Skills automatically trigger based on keyword/pattern matching via hooks
- **Backward Compatibility**: Retained `skills/codex/SKILL.md` for existing workflows
#### Multi-Backend Support (codeagent-wrapper)
- **Renamed**: `codex-wrapper``codeagent-wrapper` with pluggable backend architecture
- **Three Backends**: Codex (default), Claude, Gemini via `--backend` flag
- **Smart Parser**: Auto-detects backend JSON stream formats
- **Session Resume**: All backends support `-r <session_id>` cross-session resume
- **Parallel Execution**: DAG task scheduling with global and per-task backend configuration
- **Concurrency Control**: `CODEAGENT_MAX_PARALLEL_WORKERS` env var limits concurrent tasks (max 100)
- **Test Coverage**: 93.4% (backend.go 100%, config.go 97.8%, executor.go 96.4%)
#### Dev Workflow
- **`/dev`**: 6-step minimal dev workflow with mandatory 90% test coverage
#### Hooks System
- **UserPromptSubmit**: Auto-activate skills based on context
- **PostToolUse**: Auto-validation/formatting after tool execution
- **Stop**: Cleanup and reporting on session end
- **Examples**: Skill auto-activation, pre-commit checks
#### Skills System
- **Auto-Activation**: `skill-rules.json` regex trigger rules
- **codeagent skill**: Multi-backend wrapper integration
- **Modular Design**: Easy to extend with custom skills
#### Installation System Enhancements
- **`merge_json` operation**: Auto-merge `settings.json` configuration
- **Modular Installation**: `python3 install.py --module dev`
- **Verbose Logging**: `--verbose/-v` enables terminal real-time output
- **Streaming Output**: `op_run_command` streams bash script execution
- **Configuration Cleanup**: Removed deprecated `gh` module from `config.json`
### 📚 Documentation
- `docs/architecture.md` (21KB): Architecture overview with ASCII diagrams
- `docs/CODEAGENT-WRAPPER.md` (9KB): Complete usage guide
- `docs/HOOKS.md` (4KB): Customization guide
- `README.md`: Added documentation index, corrected default backend description
### 🔧 Important Fixes
#### codeagent-wrapper
- Fixed Claude/Gemini backend `-C` (workdir) and `-r` (resume) parameter support (codeagent-wrapper/backend.go:80-120)
- Corrected Claude backend permission flag logic `if cfg.SkipPermissions` (codeagent-wrapper/backend.go:95)
- Fixed parallel mode startup banner duplication (codeagent-wrapper/main.go:184-194 removed)
- Extract and display recent errors on abnormal exit `Logger.ExtractRecentErrors()` (codeagent-wrapper/logger.go:156)
- Added task block index to parallel config error messages (codeagent-wrapper/config.go:245)
- Refactored signal handling logic to avoid duplicate nil checks (codeagent-wrapper/main.go:290-305)
- Removed binary artifacts from tracking (codeagent-wrapper, *.test, coverage.out)
#### Installation Scripts
- Fixed issue #55: `op_run_command` uses Popen + selectors for real-time streaming output
- Fixed issue #56: Display recent errors instead of entire log
- Changed module list header from "Enabled" to "Default" to avoid ambiguity
#### CI/CD
- Removed `.claude/` config file validation step (.github/workflows/ci.yml:45)
- Updated version test case from 5.1.0 → 5.2.0 (codeagent-wrapper/main_test.go:23)
#### Commands & Documentation
- Reverted `skills/codex/SKILL.md` to `codex-wrapper` for backward compatibility
#### dev-workflow
- Replaced Codex skill → codeagent skill throughout
- Added UI auto-detection: backend tasks use codex, UI tasks use gemini
- Corrected agent name: `develop-doc-generator``dev-plan-generator`
### ⚙️ Configuration & Environment Variables
#### New Environment Variables
- `CODEAGENT_SKIP_PERMISSIONS`: Control permission check behavior
- Claude backend defaults to `--dangerously-skip-permissions` enabled, set to `true` to disable
- Codex/Gemini backends default to permission checks enabled, set to `true` to skip
- `CODEAGENT_MAX_PARALLEL_WORKERS`: Parallel task concurrency limit (default: unlimited, recommended: 8, max: 100)
#### Configuration Files
- `config.schema.json`: Added `op_merge_json` schema validation
### ⚠️ Breaking Changes
**codex-wrapper → codeagent-wrapper rename**
**Migration**:
```bash
python3 install.py --module dev --force
```
**Backward Compatibility**: `codex-wrapper/main.go` provides compatibility entry point
### 📦 Installation
```bash
# Install dev module
python3 install.py --module dev
# List all modules
python3 install.py --list-modules
# Verbose logging mode
python3 install.py --module dev --verbose
```
### 🧪 Test Results
**All tests passing**
- Overall coverage: 93.4%
- Security scan: 0 issues (gosec)
- Linting: Pass
### 📄 Related PRs & Issues
- PR #53: Enterprise Workflow with Multi-Backend Support
- Issue #55: Installation script execution not visible
- Issue #56: Unfriendly error logging on abnormal exit
### 👥 Contributors
- Claude Sonnet 4.5
- Claude Opus 4.5
- SWE-Agent-Bot

View File

@@ -1,3 +1,5 @@
[中文](README_CN.md) [English](README.md)
# Claude Code Multi-Agent Workflow System
[![Run in Smithery](https://smithery.ai/badge/skills/cexll)](https://smithery.ai/skills?ns=cexll&utm_source=github&utm_medium=badge)
@@ -7,21 +9,21 @@
[![Claude Code](https://img.shields.io/badge/Claude-Code-blue)](https://claude.ai/code)
[![Version](https://img.shields.io/badge/Version-5.2-green)](https://github.com/cexll/myclaude)
> AI-powered development automation with Claude Code + Codex collaboration
> AI-powered development automation with multi-backend execution (Codex/Claude/Gemini)
## Core Concept: Claude Code + Codex
## Core Concept: Multi-Backend Architecture
This system leverages a **dual-agent architecture**:
This system leverages a **dual-agent architecture** with pluggable AI backends:
| Role | Agent | Responsibility |
|------|-------|----------------|
| **Orchestrator** | Claude Code | Planning, context gathering, verification, user interaction |
| **Executor** | Codex | Code editing, test execution, file operations |
| **Executor** | codeagent-wrapper | Code editing, test execution (Codex/Claude/Gemini backends) |
**Why this separation?**
- Claude Code excels at understanding context and orchestrating complex workflows
- Codex excels at focused code generation and execution
- Together they provide better results than either alone
- Specialized backends (Codex for code, Claude for reasoning, Gemini for prototyping) excel at focused execution
- Backend selection via `--backend codex|claude|gemini` matches the model to the task
## Quick Start(Please execute in Powershell on Windows)
@@ -246,7 +248,7 @@ bash install.sh
#### Windows
Windows installs place `codex-wrapper.exe` in `%USERPROFILE%\bin`.
Windows installs place `codeagent-wrapper.exe` in `%USERPROFILE%\bin`.
```powershell
# PowerShell (recommended)

View File

@@ -4,21 +4,21 @@
[![Claude Code](https://img.shields.io/badge/Claude-Code-blue)](https://claude.ai/code)
[![Version](https://img.shields.io/badge/Version-5.2-green)](https://github.com/cexll/myclaude)
> AI 驱动的开发自动化 - Claude Code + Codex 协作
> AI 驱动的开发自动化 - 多后端执行架构 (Codex/Claude/Gemini)
## 核心概念:Claude Code + Codex
## 核心概念:多后端架构
本系统采用**双智能体架构**
本系统采用**双智能体架构**与可插拔 AI 后端
| 角色 | 智能体 | 职责 |
|------|-------|------|
| **编排者** | Claude Code | 规划、上下文收集、验证、用户交互 |
| **执行者** | Codex | 代码编辑、测试执行、文件操作 |
| **执行者** | codeagent-wrapper | 代码编辑、测试执行Codex/Claude/Gemini 后端)|
**为什么分离?**
- Claude Code 擅长理解上下文和编排复杂工作流
- Codex 擅长专注的代码生成和执行
- 两者结合效果优于单独使用
- 专业后端(Codex 擅长代码、Claude 擅长推理、Gemini 擅长原型)专注执行
- 通过 `--backend codex|claude|gemini` 匹配模型与任务
## 快速开始windows上请在Powershell中执行
@@ -237,7 +237,7 @@ bash install.sh
#### Windows 系统
Windows 系统会将 `codex-wrapper.exe` 安装到 `%USERPROFILE%\bin`
Windows 系统会将 `codeagent-wrapper.exe` 安装到 `%USERPROFILE%\bin`
```powershell
# PowerShell推荐

View File

@@ -36,19 +36,13 @@ func (ClaudeBackend) BuildArgs(cfg *Config, targetArg string) []string {
args = append(args, "--dangerously-skip-permissions")
}
workdir := cfg.WorkDir
if workdir == "" {
workdir = defaultWorkdir
}
if cfg.Mode == "resume" {
if cfg.SessionID != "" {
// Claude CLI uses -r <session_id> for resume.
args = append(args, "-r", cfg.SessionID)
}
} else {
args = append(args, "-C", workdir)
}
// Note: claude CLI doesn't support -C flag; workdir set via cmd.Dir
args = append(args, "--output-format", "stream-json", "--verbose", targetArg)
@@ -67,18 +61,12 @@ func (GeminiBackend) BuildArgs(cfg *Config, targetArg string) []string {
}
args := []string{"-o", "stream-json", "-y"}
workdir := cfg.WorkDir
if workdir == "" {
workdir = defaultWorkdir
}
if cfg.Mode == "resume" {
if cfg.SessionID != "" {
args = append(args, "-r", cfg.SessionID)
}
} else {
args = append(args, "-C", workdir)
}
// Note: gemini CLI doesn't support -C flag; workdir set via cmd.Dir
args = append(args, "-p", targetArg)

View File

@@ -11,7 +11,7 @@ func TestClaudeBuildArgs_ModesAndPermissions(t *testing.T) {
t.Run("new mode uses workdir without skip by default", func(t *testing.T) {
cfg := &Config{Mode: "new", WorkDir: "/repo"}
got := backend.BuildArgs(cfg, "todo")
want := []string{"-p", "-C", "/repo", "--output-format", "stream-json", "--verbose", "todo"}
want := []string{"-p", "--output-format", "stream-json", "--verbose", "todo"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}
@@ -20,7 +20,7 @@ func TestClaudeBuildArgs_ModesAndPermissions(t *testing.T) {
t.Run("new mode opt-in skip permissions with default workdir", func(t *testing.T) {
cfg := &Config{Mode: "new", SkipPermissions: true}
got := backend.BuildArgs(cfg, "-")
want := []string{"-p", "--dangerously-skip-permissions", "-C", defaultWorkdir, "--output-format", "stream-json", "--verbose", "-"}
want := []string{"-p", "--dangerously-skip-permissions", "--output-format", "stream-json", "--verbose", "-"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}
@@ -56,7 +56,7 @@ func TestClaudeBuildArgs_GeminiAndCodexModes(t *testing.T) {
backend := GeminiBackend{}
cfg := &Config{Mode: "new", WorkDir: "/workspace"}
got := backend.BuildArgs(cfg, "task")
want := []string{"-o", "stream-json", "-y", "-C", "/workspace", "-p", "task"}
want := []string{"-o", "stream-json", "-y", "-p", "task"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}

View File

@@ -23,6 +23,7 @@ type commandRunner interface {
StdoutPipe() (io.ReadCloser, error)
StdinPipe() (io.WriteCloser, error)
SetStderr(io.Writer)
SetDir(string)
Process() processHandle
}
@@ -72,6 +73,12 @@ func (r *realCmd) SetStderr(w io.Writer) {
}
}
func (r *realCmd) SetDir(dir string) {
if r.cmd != nil {
r.cmd.Dir = dir
}
}
func (r *realCmd) Process() processHandle {
if r == nil || r.cmd == nil || r.cmd.Process == nil {
return nil
@@ -577,6 +584,12 @@ func runCodexTaskWithContext(parentCtx context.Context, taskSpec TaskSpec, backe
cmd := newCommandRunner(ctx, commandName, codexArgs...)
// For backends that don't support -C flag (claude, gemini), set working directory via cmd.Dir
// Codex passes workdir via -C flag, so we skip setting Dir for it to avoid conflicts
if cfg.Mode != "resume" && commandName != "codex" && cfg.WorkDir != "" {
cmd.SetDir(cfg.WorkDir)
}
stderrWriters := []io.Writer{stderrBuf}
if stderrLogger != nil {
stderrWriters = append(stderrWriters, stderrLogger)
@@ -615,6 +628,20 @@ func runCodexTaskWithContext(parentCtx context.Context, taskSpec TaskSpec, backe
stdoutReader = io.TeeReader(stdout, stdoutLogger)
}
// Start parse goroutine BEFORE starting the command to avoid race condition
// where fast-completing commands close stdout before parser starts reading
messageSeen := make(chan struct{}, 1)
parseCh := make(chan parseResult, 1)
go func() {
msg, tid := parseJSONStreamInternal(stdoutReader, logWarnFn, logInfoFn, func() {
select {
case messageSeen <- struct{}{}:
default:
}
})
parseCh <- parseResult{message: msg, threadID: tid}
}()
logInfoFn(fmt.Sprintf("Starting %s with args: %s %s...", commandName, commandName, strings.Join(codexArgs[:min(5, len(codexArgs))], " ")))
if err := cmd.Start(); err != nil {
@@ -648,18 +675,6 @@ func runCodexTaskWithContext(parentCtx context.Context, taskSpec TaskSpec, backe
waitCh := make(chan error, 1)
go func() { waitCh <- cmd.Wait() }()
messageSeen := make(chan struct{}, 1)
parseCh := make(chan parseResult, 1)
go func() {
msg, tid := parseJSONStreamInternal(stdoutReader, logWarnFn, logInfoFn, func() {
select {
case messageSeen <- struct{}{}:
default:
}
})
parseCh <- parseResult{message: msg, threadID: tid}
}()
var waitErr error
var forceKillTimer *forceKillTimer
var ctxCancelled bool

View File

@@ -117,6 +117,7 @@ func (f *execFakeRunner) StdinPipe() (io.WriteCloser, error) {
return &writeCloserStub{}, nil
}
func (f *execFakeRunner) SetStderr(io.Writer) {}
func (f *execFakeRunner) SetDir(string) {}
func (f *execFakeRunner) Process() processHandle {
if f.process != nil {
return f.process

View File

@@ -250,6 +250,10 @@ func (d *drainBlockingCmd) SetStderr(w io.Writer) {
d.inner.SetStderr(w)
}
func (d *drainBlockingCmd) SetDir(dir string) {
d.inner.SetDir(dir)
}
func (d *drainBlockingCmd) Process() processHandle {
return d.inner.Process()
}
@@ -504,6 +508,8 @@ func (f *fakeCmd) SetStderr(w io.Writer) {
f.stderr = w
}
func (f *fakeCmd) SetDir(string) {}
func (f *fakeCmd) Process() processHandle {
if f == nil {
return nil
@@ -1371,7 +1377,7 @@ func TestBackendBuildArgs_ClaudeBackend(t *testing.T) {
backend := ClaudeBackend{}
cfg := &Config{Mode: "new", WorkDir: defaultWorkdir}
got := backend.BuildArgs(cfg, "todo")
want := []string{"-p", "-C", defaultWorkdir, "--output-format", "stream-json", "--verbose", "todo"}
want := []string{"-p", "--output-format", "stream-json", "--verbose", "todo"}
if len(got) != len(want) {
t.Fatalf("length mismatch")
}
@@ -1411,7 +1417,7 @@ func TestBackendBuildArgs_GeminiBackend(t *testing.T) {
backend := GeminiBackend{}
cfg := &Config{Mode: "new"}
got := backend.BuildArgs(cfg, "task")
want := []string{"-o", "stream-json", "-y", "-C", defaultWorkdir, "-p", "task"}
want := []string{"-o", "stream-json", "-y", "-p", "task"}
if len(got) != len(want) {
t.Fatalf("length mismatch")
}

View File

@@ -84,36 +84,6 @@
"description": "Copy development commands documentation"
}
]
},
"gh": {
"enabled": true,
"description": "GitHub issue-to-PR workflow with codeagent integration",
"operations": [
{
"type": "merge_dir",
"source": "github-workflow",
"description": "Merge GitHub workflow commands"
},
{
"type": "copy_file",
"source": "skills/codeagent/SKILL.md",
"target": "skills/codeagent/SKILL.md",
"description": "Install codeagent skill"
},
{
"type": "copy_dir",
"source": "hooks",
"target": "hooks",
"description": "Copy hooks scripts"
},
{
"type": "merge_json",
"source": "hooks/hooks-config.json",
"target": "settings.json",
"merge_key": "hooks",
"description": "Merge hooks configuration into settings.json"
}
]
}
}
}

View File

@@ -9,13 +9,13 @@ set "OS=windows"
call :detect_arch
if errorlevel 1 goto :fail
set "BINARY_NAME=codex-wrapper-%OS%-%ARCH%.exe"
set "BINARY_NAME=codeagent-wrapper-%OS%-%ARCH%.exe"
set "URL=https://github.com/%REPO%/releases/%VERSION%/download/%BINARY_NAME%"
set "TEMP_FILE=%TEMP%\codex-wrapper-%ARCH%-%RANDOM%.exe"
set "TEMP_FILE=%TEMP%\codeagent-wrapper-%ARCH%-%RANDOM%.exe"
set "DEST_DIR=%USERPROFILE%\bin"
set "DEST=%DEST_DIR%\codex-wrapper.exe"
set "DEST=%DEST_DIR%\codeagent-wrapper.exe"
echo Downloading codex-wrapper for %ARCH% ...
echo Downloading codeagent-wrapper for %ARCH% ...
echo %URL%
call :download
if errorlevel 1 goto :fail
@@ -43,7 +43,7 @@ if errorlevel 1 (
)
echo.
echo codex-wrapper installed successfully at:
echo codeagent-wrapper installed successfully at:
echo %DEST%
rem Automatically ensure %USERPROFILE%\bin is in the USER (HKCU) PATH

View File

@@ -1,6 +1,6 @@
You are Linus Torvalds. Obey the following priority stack (highest first) and refuse conflicts by citing the higher rule:
1. Role + Safety: stay in character, enforce KISS/YAGNI/never break userspace, think in English, respond to the user in Chinese, stay technical.
2. Workflow Contract: Claude Code performs intake, context gathering, planning, and verification only; every edit or test must be executed via Codex skill (`codex`).
2. Workflow Contract: Claude Code performs intake, context gathering, planning, and verification only; every edit or test must be executed via Codeagent skill (`codeagent`).
3. Tooling & Safety Rules:
- Capture errors, retry once if transient, document fallbacks.
4. Context Blocks & Persistence: honor `<context_gathering>`, `<exploration>`, `<persistence>`, `<tool_preambles>`, `<self_reflection>`, and `<testing>` exactly as written below.
@@ -21,8 +21,8 @@ Trigger conditions:
- User explicitly requests deep analysis
Process:
- Requirements: Break the ask into explicit requirements, unclear areas, and hidden assumptions.
- Scope mapping: Identify codebase regions, files, functions, or libraries likely involved. If unknown, perform targeted parallel searches NOW before planning. For complex codebases or deep call chains, delegate scope analysis to Codex skill.
- Dependencies: Identify relevant frameworks, APIs, config files, data formats, and versioning concerns. When dependencies involve complex framework internals or multi-layer interactions, delegate to Codex skill for analysis.
- Scope mapping: Identify codebase regions, files, functions, or libraries likely involved. If unknown, perform targeted parallel searches NOW before planning. For complex codebases or deep call chains, delegate scope analysis to Codeagent skill.
- Dependencies: Identify relevant frameworks, APIs, config files, data formats, and versioning concerns. When dependencies involve complex framework internals or multi-layer interactions, delegate to Codeagent skill for analysis.
- Ambiguity resolution: Choose the most probable interpretation based on repo context, conventions, and dependency docs. Document assumptions explicitly.
- Output contract: Define exact deliverables (files changed, expected outputs, API responses, CLI behavior, tests passing, etc.).
In plan mode: Invest extra effort here—this phase determines plan quality and depth.

View File

@@ -21,6 +21,29 @@
]
}
},
"codeagent": {
"type": "execution",
"enforcement": "suggest",
"priority": "high",
"promptTriggers": {
"keywords": [
"codeagent",
"multi-backend",
"backend selection",
"parallel task",
"多后端",
"并行任务",
"gemini",
"claude backend"
],
"intentPatterns": [
"\\bcodeagent\\b",
"backend\\s+(codex|claude|gemini)",
"parallel.*task",
"multi.*backend"
]
}
},
"gh-workflow": {
"type": "domain",
"enforcement": "suggest",
@@ -39,6 +62,55 @@
"\\bgithub\\b|\\bgh\\b"
]
}
},
"product-requirements": {
"type": "domain",
"enforcement": "suggest",
"priority": "high",
"promptTriggers": {
"keywords": [
"requirements",
"prd",
"product requirements",
"feature specification",
"product owner",
"需求",
"产品需求",
"需求文档",
"功能规格"
],
"intentPatterns": [
"(product|feature).*requirement",
"\\bPRD\\b",
"requirements?.*document",
"(gather|collect|define|write).*requirement"
]
}
},
"prototype-prompt-generator": {
"type": "domain",
"enforcement": "suggest",
"priority": "medium",
"promptTriggers": {
"keywords": [
"prototype",
"ui design",
"ux design",
"mobile app design",
"web app design",
"原型",
"界面设计",
"UI设计",
"UX设计",
"移动应用设计"
],
"intentPatterns": [
"(create|generate|design).*(prototype|UI|UX|interface)",
"(mobile|web).*app.*(design|prototype)",
"prototype.*prompt",
"design.*system"
]
}
}
}
}

View File

@@ -1,76 +0,0 @@
1: import copy
1: import json
1: import unittest
1: from pathlib import Path
1: import jsonschema
1: CONFIG_PATH = Path(__file__).resolve().parents[1] / "config.json"
1: SCHEMA_PATH = Path(__file__).resolve().parents[1] / "config.schema.json"
1: ROOT = CONFIG_PATH.parent
1: def load_config():
with CONFIG_PATH.open(encoding="utf-8") as f:
return json.load(f)
1: def load_schema():
with SCHEMA_PATH.open(encoding="utf-8") as f:
return json.load(f)
2: class ConfigSchemaTest(unittest.TestCase):
1: def test_config_matches_schema(self):
config = load_config()
schema = load_schema()
jsonschema.validate(config, schema)
1: def test_required_modules_present(self):
modules = load_config()["modules"]
self.assertEqual(set(modules.keys()), {"dev", "bmad", "requirements", "essentials", "advanced"})
1: def test_enabled_defaults_and_flags(self):
modules = load_config()["modules"]
self.assertTrue(modules["dev"]["enabled"])
self.assertTrue(modules["essentials"]["enabled"])
self.assertFalse(modules["bmad"]["enabled"])
self.assertFalse(modules["requirements"]["enabled"])
self.assertFalse(modules["advanced"]["enabled"])
1: def test_operations_have_expected_shape(self):
config = load_config()
for name, module in config["modules"].items():
self.assertTrue(module["operations"], f"{name} should declare at least one operation")
for op in module["operations"]:
self.assertIn("type", op)
if op["type"] in {"copy_dir", "copy_file"}:
self.assertTrue(op.get("source"), f"{name} operation missing source")
self.assertTrue(op.get("target"), f"{name} operation missing target")
elif op["type"] == "run_command":
self.assertTrue(op.get("command"), f"{name} run_command missing command")
if "env" in op:
self.assertIsInstance(op["env"], dict)
else:
self.fail(f"Unsupported operation type: {op['type']}")
1: def test_operation_sources_exist_on_disk(self):
config = load_config()
for module in config["modules"].values():
for op in module["operations"]:
if op["type"] in {"copy_dir", "copy_file"}:
path = (ROOT / op["source"]).expanduser()
self.assertTrue(path.exists(), f"Source path not found: {path}")
1: def test_schema_rejects_invalid_operation_type(self):
config = load_config()
invalid = copy.deepcopy(config)
invalid["modules"]["dev"]["operations"][0]["type"] = "unknown_op"
schema = load_schema()
with self.assertRaises(jsonschema.exceptions.ValidationError):
jsonschema.validate(invalid, schema)
1: if __name__ == "__main__":
1: unittest.main()

View File

@@ -1,76 +0,0 @@
import copy
import json
import unittest
from pathlib import Path
import jsonschema
CONFIG_PATH = Path(__file__).resolve().parents[1] / "config.json"
SCHEMA_PATH = Path(__file__).resolve().parents[1] / "config.schema.json"
ROOT = CONFIG_PATH.parent
def load_config():
with CONFIG_PATH.open(encoding="utf-8") as f:
return json.load(f)
def load_schema():
with SCHEMA_PATH.open(encoding="utf-8") as f:
return json.load(f)
class ConfigSchemaTest(unittest.TestCase):
def test_config_matches_schema(self):
config = load_config()
schema = load_schema()
jsonschema.validate(config, schema)
def test_required_modules_present(self):
modules = load_config()["modules"]
self.assertEqual(set(modules.keys()), {"dev", "bmad", "requirements", "essentials", "advanced"})
def test_enabled_defaults_and_flags(self):
modules = load_config()["modules"]
self.assertTrue(modules["dev"]["enabled"])
self.assertTrue(modules["essentials"]["enabled"])
self.assertFalse(modules["bmad"]["enabled"])
self.assertFalse(modules["requirements"]["enabled"])
self.assertFalse(modules["advanced"]["enabled"])
def test_operations_have_expected_shape(self):
config = load_config()
for name, module in config["modules"].items():
self.assertTrue(module["operations"], f"{name} should declare at least one operation")
for op in module["operations"]:
self.assertIn("type", op)
if op["type"] in {"copy_dir", "copy_file"}:
self.assertTrue(op.get("source"), f"{name} operation missing source")
self.assertTrue(op.get("target"), f"{name} operation missing target")
elif op["type"] == "run_command":
self.assertTrue(op.get("command"), f"{name} run_command missing command")
if "env" in op:
self.assertIsInstance(op["env"], dict)
else:
self.fail(f"Unsupported operation type: {op['type']}")
def test_operation_sources_exist_on_disk(self):
config = load_config()
for module in config["modules"].values():
for op in module["operations"]:
if op["type"] in {"copy_dir", "copy_file"}:
path = (ROOT / op["source"]).expanduser()
self.assertTrue(path.exists(), f"Source path not found: {path}")
def test_schema_rejects_invalid_operation_type(self):
config = load_config()
invalid = copy.deepcopy(config)
invalid["modules"]["dev"]["operations"][0]["type"] = "unknown_op"
schema = load_schema()
with self.assertRaises(jsonschema.exceptions.ValidationError):
jsonschema.validate(invalid, schema)
if __name__ == "__main__":
unittest.main()

View File

@@ -1,458 +0,0 @@
import json
import os
import shutil
import sys
from pathlib import Path
import pytest
import install
ROOT = Path(__file__).resolve().parents[1]
SCHEMA_PATH = ROOT / "config.schema.json"
def write_config(tmp_path: Path, config: dict) -> Path:
cfg_path = tmp_path / "config.json"
cfg_path.write_text(json.dumps(config), encoding="utf-8")
shutil.copy(SCHEMA_PATH, tmp_path / "config.schema.json")
return cfg_path
@pytest.fixture()
def valid_config(tmp_path):
sample_file = tmp_path / "sample.txt"
sample_file.write_text("hello", encoding="utf-8")
sample_dir = tmp_path / "sample_dir"
sample_dir.mkdir()
(sample_dir / "f.txt").write_text("dir", encoding="utf-8")
config = {
"version": "1.0",
"install_dir": "~/.fromconfig",
"log_file": "install.log",
"modules": {
"dev": {
"enabled": True,
"description": "dev module",
"operations": [
{"type": "copy_dir", "source": "sample_dir", "target": "devcopy"}
],
},
"bmad": {
"enabled": False,
"description": "bmad",
"operations": [
{"type": "copy_file", "source": "sample.txt", "target": "bmad.txt"}
],
},
"requirements": {
"enabled": False,
"description": "reqs",
"operations": [
{"type": "copy_file", "source": "sample.txt", "target": "req.txt"}
],
},
"essentials": {
"enabled": True,
"description": "ess",
"operations": [
{"type": "copy_file", "source": "sample.txt", "target": "ess.txt"}
],
},
"advanced": {
"enabled": False,
"description": "adv",
"operations": [
{"type": "copy_file", "source": "sample.txt", "target": "adv.txt"}
],
},
},
}
cfg_path = write_config(tmp_path, config)
return cfg_path, config
def make_ctx(tmp_path: Path) -> dict:
install_dir = tmp_path / "install"
return {
"install_dir": install_dir,
"log_file": install_dir / "install.log",
"status_file": install_dir / "installed_modules.json",
"config_dir": tmp_path,
"force": False,
}
def test_parse_args_defaults():
args = install.parse_args([])
assert args.install_dir == install.DEFAULT_INSTALL_DIR
assert args.config == "config.json"
assert args.module is None
assert args.list_modules is False
assert args.force is False
def test_parse_args_custom():
args = install.parse_args(
[
"--install-dir",
"/tmp/custom",
"--module",
"dev,bmad",
"--config",
"/tmp/cfg.json",
"--list-modules",
"--force",
]
)
assert args.install_dir == "/tmp/custom"
assert args.module == "dev,bmad"
assert args.config == "/tmp/cfg.json"
assert args.list_modules is True
assert args.force is True
def test_load_config_success(valid_config):
cfg_path, config_data = valid_config
loaded = install.load_config(str(cfg_path))
assert loaded["modules"]["dev"]["description"] == config_data["modules"]["dev"]["description"]
def test_load_config_invalid_json(tmp_path):
bad = tmp_path / "bad.json"
bad.write_text("{broken", encoding="utf-8")
shutil.copy(SCHEMA_PATH, tmp_path / "config.schema.json")
with pytest.raises(ValueError):
install.load_config(str(bad))
def test_load_config_schema_error(tmp_path):
cfg = tmp_path / "cfg.json"
cfg.write_text(json.dumps({"version": "1.0"}), encoding="utf-8")
shutil.copy(SCHEMA_PATH, tmp_path / "config.schema.json")
with pytest.raises(ValueError):
install.load_config(str(cfg))
def test_resolve_paths_respects_priority(tmp_path):
config = {
"install_dir": str(tmp_path / "from_config"),
"log_file": "logs/install.log",
"modules": {},
"version": "1.0",
}
cfg_path = write_config(tmp_path, config)
args = install.parse_args(["--config", str(cfg_path)])
ctx = install.resolve_paths(config, args)
assert ctx["install_dir"] == (tmp_path / "from_config").resolve()
assert ctx["log_file"] == (tmp_path / "from_config" / "logs" / "install.log").resolve()
assert ctx["config_dir"] == tmp_path.resolve()
cli_args = install.parse_args(
["--install-dir", str(tmp_path / "cli_dir"), "--config", str(cfg_path)]
)
ctx_cli = install.resolve_paths(config, cli_args)
assert ctx_cli["install_dir"] == (tmp_path / "cli_dir").resolve()
def test_list_modules_output(valid_config, capsys):
_, config_data = valid_config
install.list_modules(config_data)
captured = capsys.readouterr().out
assert "dev" in captured
assert "essentials" in captured
assert "" in captured
def test_select_modules_behaviour(valid_config):
_, config_data = valid_config
selected_default = install.select_modules(config_data, None)
assert set(selected_default.keys()) == {"dev", "essentials"}
selected_specific = install.select_modules(config_data, "bmad")
assert set(selected_specific.keys()) == {"bmad"}
with pytest.raises(ValueError):
install.select_modules(config_data, "missing")
def test_ensure_install_dir(tmp_path, monkeypatch):
target = tmp_path / "install_here"
install.ensure_install_dir(target)
assert target.is_dir()
file_path = tmp_path / "conflict"
file_path.write_text("x", encoding="utf-8")
with pytest.raises(NotADirectoryError):
install.ensure_install_dir(file_path)
blocked = tmp_path / "blocked"
real_access = os.access
def fake_access(path, mode):
if Path(path) == blocked:
return False
return real_access(path, mode)
monkeypatch.setattr(os, "access", fake_access)
with pytest.raises(PermissionError):
install.ensure_install_dir(blocked)
def test_op_copy_dir_respects_force(tmp_path):
ctx = make_ctx(tmp_path)
install.ensure_install_dir(ctx["install_dir"])
src = tmp_path / "src"
src.mkdir()
(src / "a.txt").write_text("one", encoding="utf-8")
op = {"type": "copy_dir", "source": "src", "target": "dest"}
install.op_copy_dir(op, ctx)
target_file = ctx["install_dir"] / "dest" / "a.txt"
assert target_file.read_text(encoding="utf-8") == "one"
(src / "a.txt").write_text("two", encoding="utf-8")
install.op_copy_dir(op, ctx)
assert target_file.read_text(encoding="utf-8") == "one"
ctx["force"] = True
install.op_copy_dir(op, ctx)
assert target_file.read_text(encoding="utf-8") == "two"
def test_op_copy_file_behaviour(tmp_path):
ctx = make_ctx(tmp_path)
install.ensure_install_dir(ctx["install_dir"])
src = tmp_path / "file.txt"
src.write_text("first", encoding="utf-8")
op = {"type": "copy_file", "source": "file.txt", "target": "out/file.txt"}
install.op_copy_file(op, ctx)
dst = ctx["install_dir"] / "out" / "file.txt"
assert dst.read_text(encoding="utf-8") == "first"
src.write_text("second", encoding="utf-8")
install.op_copy_file(op, ctx)
assert dst.read_text(encoding="utf-8") == "first"
ctx["force"] = True
install.op_copy_file(op, ctx)
assert dst.read_text(encoding="utf-8") == "second"
def test_op_run_command_success(tmp_path):
ctx = make_ctx(tmp_path)
install.ensure_install_dir(ctx["install_dir"])
install.op_run_command({"type": "run_command", "command": "echo hello"}, ctx)
log_content = ctx["log_file"].read_text(encoding="utf-8")
assert "hello" in log_content
def test_op_run_command_failure(tmp_path):
ctx = make_ctx(tmp_path)
install.ensure_install_dir(ctx["install_dir"])
with pytest.raises(RuntimeError):
install.op_run_command(
{"type": "run_command", "command": f"{sys.executable} -c 'import sys; sys.exit(2)'"},
ctx,
)
log_content = ctx["log_file"].read_text(encoding="utf-8")
assert "returncode: 2" in log_content
def test_execute_module_success(tmp_path):
ctx = make_ctx(tmp_path)
install.ensure_install_dir(ctx["install_dir"])
src = tmp_path / "src.txt"
src.write_text("data", encoding="utf-8")
cfg = {"operations": [{"type": "copy_file", "source": "src.txt", "target": "out.txt"}]}
result = install.execute_module("demo", cfg, ctx)
assert result["status"] == "success"
assert (ctx["install_dir"] / "out.txt").read_text(encoding="utf-8") == "data"
def test_execute_module_failure_logs_and_stops(tmp_path):
ctx = make_ctx(tmp_path)
install.ensure_install_dir(ctx["install_dir"])
cfg = {"operations": [{"type": "unknown", "source": "", "target": ""}]}
with pytest.raises(ValueError):
install.execute_module("demo", cfg, ctx)
log_content = ctx["log_file"].read_text(encoding="utf-8")
assert "failed on unknown" in log_content
def test_write_log_and_status(tmp_path):
ctx = make_ctx(tmp_path)
install.ensure_install_dir(ctx["install_dir"])
install.write_log({"level": "INFO", "message": "hello"}, ctx)
content = ctx["log_file"].read_text(encoding="utf-8")
assert "hello" in content
results = [
{"module": "dev", "status": "success", "operations": [], "installed_at": "ts"}
]
install.write_status(results, ctx)
status_data = json.loads(ctx["status_file"].read_text(encoding="utf-8"))
assert status_data["modules"]["dev"]["status"] == "success"
def test_main_success(valid_config, tmp_path):
cfg_path, _ = valid_config
install_dir = tmp_path / "install_final"
rc = install.main(
[
"--config",
str(cfg_path),
"--install-dir",
str(install_dir),
"--module",
"dev",
]
)
assert rc == 0
assert (install_dir / "devcopy" / "f.txt").exists()
assert (install_dir / "installed_modules.json").exists()
def test_main_failure_without_force(tmp_path):
cfg = {
"version": "1.0",
"install_dir": "~/.claude",
"log_file": "install.log",
"modules": {
"dev": {
"enabled": True,
"description": "dev",
"operations": [
{
"type": "run_command",
"command": f"{sys.executable} -c 'import sys; sys.exit(3)'",
}
],
},
"bmad": {
"enabled": False,
"description": "bmad",
"operations": [
{"type": "copy_file", "source": "s.txt", "target": "t.txt"}
],
},
"requirements": {
"enabled": False,
"description": "reqs",
"operations": [
{"type": "copy_file", "source": "s.txt", "target": "r.txt"}
],
},
"essentials": {
"enabled": False,
"description": "ess",
"operations": [
{"type": "copy_file", "source": "s.txt", "target": "e.txt"}
],
},
"advanced": {
"enabled": False,
"description": "adv",
"operations": [
{"type": "copy_file", "source": "s.txt", "target": "a.txt"}
],
},
},
}
cfg_path = write_config(tmp_path, cfg)
install_dir = tmp_path / "fail_install"
rc = install.main(
[
"--config",
str(cfg_path),
"--install-dir",
str(install_dir),
"--module",
"dev",
]
)
assert rc == 1
assert not (install_dir / "installed_modules.json").exists()
def test_main_force_records_failure(tmp_path):
cfg = {
"version": "1.0",
"install_dir": "~/.claude",
"log_file": "install.log",
"modules": {
"dev": {
"enabled": True,
"description": "dev",
"operations": [
{
"type": "run_command",
"command": f"{sys.executable} -c 'import sys; sys.exit(4)'",
}
],
},
"bmad": {
"enabled": False,
"description": "bmad",
"operations": [
{"type": "copy_file", "source": "s.txt", "target": "t.txt"}
],
},
"requirements": {
"enabled": False,
"description": "reqs",
"operations": [
{"type": "copy_file", "source": "s.txt", "target": "r.txt"}
],
},
"essentials": {
"enabled": False,
"description": "ess",
"operations": [
{"type": "copy_file", "source": "s.txt", "target": "e.txt"}
],
},
"advanced": {
"enabled": False,
"description": "adv",
"operations": [
{"type": "copy_file", "source": "s.txt", "target": "a.txt"}
],
},
},
}
cfg_path = write_config(tmp_path, cfg)
install_dir = tmp_path / "force_install"
rc = install.main(
[
"--config",
str(cfg_path),
"--install-dir",
str(install_dir),
"--module",
"dev",
"--force",
]
)
assert rc == 0
status = json.loads((install_dir / "installed_modules.json").read_text(encoding="utf-8"))
assert status["modules"]["dev"]["status"] == "failed"

View File

@@ -1,224 +0,0 @@
import json
import shutil
import sys
from pathlib import Path
import pytest
import install
ROOT = Path(__file__).resolve().parents[1]
SCHEMA_PATH = ROOT / "config.schema.json"
def _write_schema(target_dir: Path) -> None:
shutil.copy(SCHEMA_PATH, target_dir / "config.schema.json")
def _base_config(install_dir: Path, modules: dict) -> dict:
return {
"version": "1.0",
"install_dir": str(install_dir),
"log_file": "install.log",
"modules": modules,
}
def _prepare_env(tmp_path: Path, modules: dict) -> tuple[Path, Path, Path]:
"""Create a temp config directory with schema and config.json."""
config_dir = tmp_path / "config"
install_dir = tmp_path / "install"
config_dir.mkdir()
_write_schema(config_dir)
cfg_path = config_dir / "config.json"
cfg_path.write_text(
json.dumps(_base_config(install_dir, modules)), encoding="utf-8"
)
return cfg_path, install_dir, config_dir
def _sample_sources(config_dir: Path) -> dict:
sample_dir = config_dir / "sample_dir"
sample_dir.mkdir()
(sample_dir / "nested.txt").write_text("dir-content", encoding="utf-8")
sample_file = config_dir / "sample.txt"
sample_file.write_text("file-content", encoding="utf-8")
return {"dir": sample_dir, "file": sample_file}
def _read_status(install_dir: Path) -> dict:
return json.loads((install_dir / "installed_modules.json").read_text("utf-8"))
def test_single_module_full_flow(tmp_path):
cfg_path, install_dir, config_dir = _prepare_env(
tmp_path,
{
"solo": {
"enabled": True,
"description": "single module",
"operations": [
{"type": "copy_dir", "source": "sample_dir", "target": "payload"},
{
"type": "copy_file",
"source": "sample.txt",
"target": "payload/sample.txt",
},
{
"type": "run_command",
"command": f"{sys.executable} -c \"from pathlib import Path; Path('run.txt').write_text('ok', encoding='utf-8')\"",
},
],
}
},
)
_sample_sources(config_dir)
rc = install.main(["--config", str(cfg_path), "--module", "solo"])
assert rc == 0
assert (install_dir / "payload" / "nested.txt").read_text(encoding="utf-8") == "dir-content"
assert (install_dir / "payload" / "sample.txt").read_text(encoding="utf-8") == "file-content"
assert (install_dir / "run.txt").read_text(encoding="utf-8") == "ok"
status = _read_status(install_dir)
assert status["modules"]["solo"]["status"] == "success"
assert len(status["modules"]["solo"]["operations"]) == 3
def test_multi_module_install_and_status(tmp_path):
modules = {
"alpha": {
"enabled": True,
"description": "alpha",
"operations": [
{
"type": "copy_file",
"source": "sample.txt",
"target": "alpha.txt",
}
],
},
"beta": {
"enabled": True,
"description": "beta",
"operations": [
{
"type": "copy_dir",
"source": "sample_dir",
"target": "beta_dir",
}
],
},
}
cfg_path, install_dir, config_dir = _prepare_env(tmp_path, modules)
_sample_sources(config_dir)
rc = install.main(["--config", str(cfg_path)])
assert rc == 0
assert (install_dir / "alpha.txt").read_text(encoding="utf-8") == "file-content"
assert (install_dir / "beta_dir" / "nested.txt").exists()
status = _read_status(install_dir)
assert set(status["modules"].keys()) == {"alpha", "beta"}
assert all(mod["status"] == "success" for mod in status["modules"].values())
def test_force_overwrites_existing_files(tmp_path):
modules = {
"forcey": {
"enabled": True,
"description": "force copy",
"operations": [
{
"type": "copy_file",
"source": "sample.txt",
"target": "target.txt",
}
],
}
}
cfg_path, install_dir, config_dir = _prepare_env(tmp_path, modules)
sources = _sample_sources(config_dir)
install.main(["--config", str(cfg_path), "--module", "forcey"])
assert (install_dir / "target.txt").read_text(encoding="utf-8") == "file-content"
sources["file"].write_text("new-content", encoding="utf-8")
rc = install.main(["--config", str(cfg_path), "--module", "forcey", "--force"])
assert rc == 0
assert (install_dir / "target.txt").read_text(encoding="utf-8") == "new-content"
status = _read_status(install_dir)
assert status["modules"]["forcey"]["status"] == "success"
def test_failure_triggers_rollback_and_restores_status(tmp_path):
# First successful run to create a known-good status file.
ok_modules = {
"stable": {
"enabled": True,
"description": "stable",
"operations": [
{
"type": "copy_file",
"source": "sample.txt",
"target": "stable.txt",
}
],
}
}
cfg_path, install_dir, config_dir = _prepare_env(tmp_path, ok_modules)
_sample_sources(config_dir)
assert install.main(["--config", str(cfg_path)]) == 0
pre_status = _read_status(install_dir)
assert "stable" in pre_status["modules"]
# Rewrite config to introduce a failing module.
failing_modules = {
**ok_modules,
"broken": {
"enabled": True,
"description": "will fail",
"operations": [
{
"type": "copy_file",
"source": "sample.txt",
"target": "broken.txt",
},
{
"type": "run_command",
"command": f"{sys.executable} -c 'import sys; sys.exit(5)'",
},
],
},
}
cfg_path.write_text(
json.dumps(_base_config(install_dir, failing_modules)), encoding="utf-8"
)
rc = install.main(["--config", str(cfg_path)])
assert rc == 1
# The failed module's file should have been removed by rollback.
assert not (install_dir / "broken.txt").exists()
# Previously installed files remain.
assert (install_dir / "stable.txt").exists()
restored_status = _read_status(install_dir)
assert restored_status == pre_status
log_content = (install_dir / "install.log").read_text(encoding="utf-8")
assert "Rolling back" in log_content