docs(scripts): document benchmark and context report

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

README.md

@@ -220,6 +220,10 @@ Existing `.agents/skills/backups/` content is moved there on the next install, u
 Codex `config.toml` synchronization merges setup-owned keys while retaining other root keys, extra fields in managed
 tables, plugin and marketplace tables, other unmanaged tables, and unowned array tables. The previous
 file is backed up when content changes. Claude `settings.json` synchronization retains unrelated settings and hooks.
+Every managed file, including `config.toml`, `settings.json`, and the manifest, is written to a temporary sibling
+file and then renamed over the target, so an interrupted run never leaves a partly written file. Each destination is
+locked through `.ai-config.lock` while it syncs; a parallel install or update fails instead of interleaving writes.
+A lock left by an exited process of the same shell on the same host is replaced automatically.
 Dry runs still rebuild `generated/`, but do not modify installation destinations.
 
 Selecting Codex also installs shared skills to `.agents/skills`. An interactive extension list includes plugins and

docs/scripts.md

@@ -22,6 +22,8 @@ Run scripts from the repository root. PowerShell and Bash commands have equivale
 | `commands/uninstall.ps1`, `commands/uninstall.sh` | Remove setup-owned configuration and extensions. |
 | `commands/doctor.ps1`, `commands/doctor.sh` | Check generated output and installed configuration. |
 | `commands/prompt-budget.ps1`, `commands/prompt-budget.sh` | Track permanent context, skill metadata, the rule catalog, and agent metadata; warn above 10,000 permanent-context tokens, fail when any token estimate exceeds its baseline by 10%, or permanent context exceeds 15,000 tokens. |
+| `commands/context-report.ps1`, `commands/context-report.sh` | Report the context installed Claude and Codex actually load, from Claude `/context`, `/skill-doctor`, and its init event, and Codex `debug prompt-input`: categories, active plugins, listed skills, installed skills missing from the listing, and skills listed without description context. `--check` fails on missing or description-less skills. No model request is made. |
+| `commands/benchmark.ps1`, `commands/benchmark.sh` | Run the ten tasks under `scripts/benchmarks/tasks/` with and without the installed setup and compare pass rate, rework turns, duration, tokens, tool calls, cost, and workflow checks (no planning ceremony for a small change, targeted verification for a bugfix, bounded delegation for a larger change). Starts real model sessions; use `--dry-run` to list runs. The Codex baseline needs one `codex login` with `CODEX_HOME` set to `--codex-baseline-home`. |
 | `commands/repository-context.ps1`, `commands/repository-context.sh` | Refresh the local repository-context cache. |
 | `commands/session-state.ps1`, `commands/session-state.sh` | Read or update the local agent session state. |
 | `commands/telemetry.ps1`, `commands/telemetry.sh` | Append a local telemetry event. |
@@ -38,7 +40,7 @@ To undo an added instruction, pass the same text to `remove-instruction` with `-
 Credential commands currently require PowerShell and Windows Credential Manager, including their Bash wrappers. Use `~/.my-ai-configuration/bin/with-credential.ps1` or `.sh` to run a tool with a stored key available only in that child process. The key must be a valid environment-variable name; its value is never written to configuration.
 Use `remove-credentials` with `-Client`/`--client` and `-Key`/`--key` to delete only that credential. The shared `with-credential` helpers remain installed for other keys.
 
-Qwen Auto selection uses `shared/retrieval/benchmark.py`; run it through `scripts/lib/semantic-retrieval.py benchmark --root . --home <home-directory>`. This warm-model check uses synthetic, repeated short documents. Its result does not measure model startup, first indexing, real code chunks, CLI latency, or memory use; do not use it as an end-to-end performance estimate.
+The A/B benchmark measures whole sessions. Qwen Auto selection uses `shared/retrieval/benchmark.py`; run it through `scripts/lib/semantic-retrieval.py benchmark --root . --home <home-directory>`. This warm-model check uses synthetic, repeated short documents. Its result does not measure model startup, first indexing, real code chunks, CLI latency, or memory use; do not use it as an end-to-end performance estimate.
 
 ## Checks and tests
 

scripts/benchmarks/ab.py

@@ -0,0 +1,270 @@
+"""Run the benchmark tasks with and without the installed setup and compare the results.
+
+Each run copies a task fixture into a fresh Git repository, gives the client the task prompt,
+then runs hidden checks the agent never saw. A failed check gets one rework turn with the check
+output. Results land in <out>/results.jsonl and <out>/summary.md.
+
+Arms:
+  setup     the installed configuration (CLAUDE.md, settings, plugins, skills, MCP servers).
+  baseline  Claude with --setting-sources "" --strict-mcp-config (no user settings, plugins,
+            user skills, CLAUDE.md, or MCP servers); Codex with CODEX_HOME set to --codex-baseline-home,
+            an empty home where you ran `codex login` once.
+"""
+import argparse
+from concurrent.futures import ThreadPoolExecutor
+import json
+import os
+from pathlib import Path
+import queue
+import re
+import shutil
+import statistics
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+
+TASKS = Path(__file__).resolve().parent / 'tasks'
+CHECK = [sys.executable, '-m', 'unittest', 'discover', '-s', 'bench_hidden', '-t', '.']
+CLAUDE_TOOLS = ['Read', 'Edit', 'Write', 'Glob', 'Grep', 'Agent', 'Skill', 'TodoWrite', 'Bash(python *)', 'Bash(git *)',
+                'Bash(ls *)', 'PowerShell(python *)', 'PowerShell(git *)', 'PowerShell(Get-ChildItem *)']
+
+
+def clean_env(**values):
+    env = {key: value for key, value in os.environ.items() if not key.startswith(('BUN_INSPECT', 'CLAUDECODE', 'CLAUDE_CODE_ENTRYPOINT'))}
+    env.update(values)
+    return env
+
+
+def kill_tree(process):
+    if os.name == 'nt':
+        subprocess.run(['taskkill', '/T', '/F', '/PID', str(process.pid)], capture_output=True, timeout=30)
+    else:
+        process.kill()
+    process.wait(timeout=30)
+
+
+def run_jsonl(command, cwd, env, timeout):
+    """Run a client, return (events, seconds, timed_out); output is read on a thread so the deadline holds."""
+    start = time.monotonic()
+    process = subprocess.Popen(command, cwd=cwd, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
+                               stderr=subprocess.DEVNULL, text=True, encoding='utf-8', errors='replace')
+    lines = queue.Queue()
+    threading.Thread(target=lambda: ([lines.put(line) for line in process.stdout], lines.put(None)), daemon=True).start()
+    events, timed_out = [], False
+    while True:
+        try:
+            line = lines.get(timeout=max(1, timeout - (time.monotonic() - start)))
+        except queue.Empty:
+            timed_out = True
+            break
+        if line is None:
+            break
+        try:
+            events.append(json.loads(line))
+        except ValueError:
+            pass
+    kill_tree(process) if process.poll() is None else process.wait()
+    return events, time.monotonic() - start, timed_out
+
+
+def claude_turn(prompt, cwd, arm, model, session, timeout):
+    command = [shutil.which('claude'), '-p', prompt, '--output-format', 'stream-json', '--verbose', '--model', model,
+               '--permission-mode', 'auto', '--allowedTools', *CLAUDE_TOOLS]
+    if arm == 'baseline':
+        command += ['--setting-sources', '', '--strict-mcp-config']
+    if session:
+        command += ['--resume', session]
+    events, seconds, timed_out = run_jsonl(command, cwd, clean_env(), timeout)
+    result = next((event for event in reversed(events) if event.get('type') == 'result'), {})
+    usage = result.get('usage', {})
+    calls = [block for event in events if event.get('type') == 'assistant'
+             for block in event.get('message', {}).get('content', []) if block.get('type') == 'tool_use']
+    commands = [str(call.get('input', {}).get('command', '')) for call in calls if call.get('name') in ('Bash', 'PowerShell')]
+    return {
+        'session': result.get('session_id') or session, 'seconds': seconds, 'timed_out': timed_out,
+        'input_tokens': usage.get('input_tokens', 0) + usage.get('cache_read_input_tokens', 0) + usage.get('cache_creation_input_tokens', 0),
+        'output_tokens': usage.get('output_tokens', 0), 'cost_usd': result.get('total_cost_usd') or 0,
+        'tool_calls': len(calls), 'agents': sum(call.get('name') == 'Agent' for call in calls),
+        'skills': [str(call.get('input', {}).get('skill', '')) for call in calls if call.get('name') == 'Skill'],
+        'commands': commands, 'final': str(result.get('result', '')),
+        'error': str(result.get('result', '') or 'no result event')[:300] if result.get('is_error') or not result else '',
+    }
+
+
+def codex_turn(prompt, cwd, arm, model, session, timeout, baseline_home):
+    command = [shutil.which('codex'), 'exec']
+    if session:
+        command += ['resume', session]
+    command += ['--json', '--skip-git-repo-check', '--sandbox', 'workspace-write']
+    if model:
+        command += ['-m', model]
+    command += ['-C', str(cwd)] if not session else []
+    command.append(prompt)
+    env = clean_env(CODEX_HOME=str(baseline_home)) if arm == 'baseline' else clean_env()
+    events, seconds, timed_out = run_jsonl(command, cwd, env, timeout)
+    items = [event.get('item', {}) for event in events if event.get('type') == 'item.completed']
+    tools = [item for item in items if item.get('type') not in ('agent_message', 'reasoning', 'error')]
+    usage = [event.get('usage', {}) for event in events if event.get('type') == 'turn.completed']
+    thread = next((event.get('thread_id') for event in events if event.get('type') == 'thread.started'), session)
+    return {
+        'session': thread, 'seconds': seconds, 'timed_out': timed_out,
+        'input_tokens': sum(entry.get('input_tokens', 0) for entry in usage),
+        'output_tokens': sum(entry.get('output_tokens', 0) for entry in usage), 'cost_usd': 0,
+        'tool_calls': len(tools), 'agents': sum(item.get('type') == 'collab_tool_call' for item in tools),
+        'skills': [], 'commands': [str(item.get('command', '')) for item in tools if item.get('type') == 'command_execution'],
+        'final': next((str(item.get('text', '')) for item in reversed(items) if item.get('type') == 'agent_message'), ''),
+        'error': next((str(event.get('message', event.get('error', '')))[:300] for event in events if event.get('type') in ('error', 'turn.failed')
+                       and not str(event.get('message', '')).startswith('Reconnecting')), '') if not usage else '',
+    }
+
+
+def check(work, task):
+    hidden = work / 'bench_hidden'
+    shutil.rmtree(hidden, ignore_errors=True)
+    shutil.copytree(task / 'hidden', hidden)
+    result = subprocess.run(CHECK, cwd=work, capture_output=True, text=True, timeout=300)
+    shutil.rmtree(hidden, ignore_errors=True)
+    return result.returncode == 0, (result.stdout + result.stderr)[-3000:]
+
+
+def workflow_violations(spec, turns):
+    skills = [skill for turn in turns for skill in turn['skills']]
+    agents = sum(turn['agents'] for turn in turns)
+    calls = sum(turn['tool_calls'] for turn in turns)
+    commands = [command for turn in turns for command in turn['commands']]
+    problems = [f'used planning skill {skill}' for skill in skills if skill in spec.get('forbid_skills', [])]
+    if agents > spec.get('max_agents', 99):
+        problems.append(f'{agents} agents > {spec["max_agents"]}')
+    if agents < spec.get('min_agents', 0):
+        problems.append(f'{agents} agents < {spec["min_agents"]}')
+    if calls > spec.get('max_tool_calls', 10 ** 6):
+        problems.append(f'{calls} tool calls > {spec["max_tool_calls"]}')
+    if spec.get('require_command') and not any(re.search(spec['require_command'], command) for command in commands):
+        problems.append(f'no command matching {spec["require_command"]}')
+    return problems
+
+
+def run_one(job, args):
+    client, arm, task, repetition = job
+    spec = json.loads((task / 'task.json').read_text(encoding='utf-8'))
+    with tempfile.TemporaryDirectory(prefix=f'ai-bench-{task.name[:2]}-', ignore_cleanup_errors=True) as directory:
+        work = Path(directory)
+        shutil.copytree(task / 'repo', work, dirs_exist_ok=True)
+        for git in (['init', '-q'], ['add', '-A'], ['-c', 'user.name=bench', '-c', 'user.email=bench@example.invalid', 'commit', '-qm', 'fixture']):
+            subprocess.run(['git', '-c', 'core.autocrlf=false', *git], cwd=work, check=True, timeout=60)
+        turns, prompt, session = [], spec['prompt'], None
+        for round_number in range(1 + args.rework):
+            if client == 'claude':
+                turn = claude_turn(prompt, work, arm, args.claude_model, session, args.timeout)
+            else:
+                turn = codex_turn(prompt, work, arm, args.codex_model, session, args.timeout, args.codex_baseline_home)
+            turns.append(turn)
+            session = turn['session']
+            if turn['error']:
+                passed, output = False, ''
+                break
+            passed, output = check(work, task)
+            if passed or turn['timed_out'] or not session:
+                break
+            prompt = f'The acceptance checks failed:\n\n{output}\n\nFix the code so they pass.'
+        if not passed and args.keep_failed:
+            shutil.copytree(work, args.out / 'failed' / f'{client}-{arm}-{task.name}-{repetition}', ignore=shutil.ignore_patterns('.git'))
+    return {
+        'client': client, 'arm': arm, 'task': task.name, 'repetition': repetition, 'passed': passed,
+        'error': next((turn['error'] for turn in turns if turn['error']), ''),
+        'first_try': passed and len(turns) == 1, 'rework_turns': len(turns) - 1,
+        'seconds': round(sum(turn['seconds'] for turn in turns), 1),
+        'input_tokens': sum(turn['input_tokens'] for turn in turns), 'output_tokens': sum(turn['output_tokens'] for turn in turns),
+        'cost_usd': round(sum(turn['cost_usd'] for turn in turns), 4), 'tool_calls': sum(turn['tool_calls'] for turn in turns),
+        'agents': sum(turn['agents'] for turn in turns), 'skills': [skill for turn in turns for skill in turn['skills']],
+        'commands': [command for turn in turns for command in turn['commands']],
+        'timed_out': any(turn['timed_out'] for turn in turns), 'workflow_kind': spec['workflow'].get('kind'),
+        'workflow_violations': workflow_violations(spec['workflow'], turns), 'check_output': '' if passed else output,
+    }
+
+
+def summarize(results):
+    errors = [result for result in results if result.get('error')]
+    results = [result for result in results if not result.get('error')]
+    lines = [f'Excluded {len(errors)} runs that ended with a client or API error (for example a usage limit).', '','| client | arm | runs | pass | first try | rework turns | median s | median input tok | median output tok | median tool calls | cost USD | workflow ok |',
+             '|---|---|---|---|---|---|---|---|---|---|---|---|']
+    groups = {}
+    for result in results:
+        groups.setdefault((result['client'], result['arm']), []).append(result)
+    for (client, arm), runs in sorted(groups.items()):
+        median = lambda key: statistics.median(run[key] for run in runs)
+        lines.append(f'| {client} | {arm} | {len(runs)} | {sum(run["passed"] for run in runs) / len(runs):.0%} | '
+                     f'{sum(run["first_try"] for run in runs) / len(runs):.0%} | {sum(run["rework_turns"] for run in runs)} | '
+                     f'{median("seconds"):.0f} | {median("input_tokens"):.0f} | {median("output_tokens"):.0f} | {median("tool_calls"):.0f} | '
+                     f'{sum(run["cost_usd"] for run in runs):.2f} | {sum(not run["workflow_violations"] for run in runs) / len(runs):.0%} |')
+    lines += ['', '## Workflow checks', '', '| client | arm | task | violations |', '|---|---|---|---|']
+    for result in sorted(results, key=lambda run: (run['client'], run['arm'], run['task'], run['repetition'])):
+        if result['workflow_violations']:
+            lines.append(f'| {result["client"]} | {result["arm"]} | {result["task"]} #{result["repetition"]} | {"; ".join(result["workflow_violations"])} |')
+    return '\n'.join(lines) + '\n'
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0], formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__)
+    parser.add_argument('--clients', nargs='+', choices=('claude', 'codex'), default=['claude', 'codex'])
+    parser.add_argument('--arms', nargs='+', choices=('setup', 'baseline'), default=['setup', 'baseline'])
+    parser.add_argument('--tasks', nargs='+', help='task directory names (default: all)')
+    parser.add_argument('--repetitions', type=int, default=3)
+    parser.add_argument('--first-repetition', type=int, default=1, help='number of the first repetition, to rerun selected repetitions')
+    parser.add_argument('--rework', type=int, default=1, help='rework turns after a failed check')
+    parser.add_argument('--jobs', type=int, default=2)
+    parser.add_argument('--timeout', type=int, default=900, help='seconds per client turn')
+    parser.add_argument('--claude-model', default='sonnet')
+    parser.add_argument('--codex-model', help='default: the CLI default, identical for both arms')
+    parser.add_argument('--codex-baseline-home', type=Path, default=Path.home() / '.my-ai-configuration/benchmark/codex-home')
+    parser.add_argument('--out', type=Path, default=Path(tempfile.gettempdir()) / f'ai-config-ab-{time.strftime("%Y%m%d-%H%M%S")}')
+    parser.add_argument('--dry-run', action='store_true', help='list the runs without starting clients')
+    parser.add_argument('--keep-failed', action='store_true', help='copy failed work trees to <out>/failed')
+    parser.add_argument('--summarize', nargs='+', type=Path, metavar='RESULTS', help='only summarize results.jsonl files; a later file replaces runs of an earlier one')
+    args = parser.parse_args()
+    if args.summarize:
+        merged = {}
+        for path in args.summarize:
+            for line in path.read_text(encoding='utf-8').splitlines():
+                result = json.loads(line)
+                key = (result['client'], result['arm'], result['task'], result['repetition'])
+                if not result.get('error') or key not in merged:
+                    merged[key] = result
+        print(summarize(list(merged.values())))
+        return 0
+    tasks = [TASKS / name for name in args.tasks] if args.tasks else sorted(path for path in TASKS.iterdir() if path.is_dir())
+    for client in args.clients:
+        if not shutil.which(client):
+            parser.error(f'{client} CLI unavailable')
+    if 'codex' in args.clients and 'baseline' in args.arms and not (args.codex_baseline_home / 'auth.json').is_file():
+        parser.error(f'Codex baseline needs a login: run CODEX_HOME="{args.codex_baseline_home}" codex login')
+    jobs = [(client, arm, task, repetition) for repetition in range(args.first_repetition, args.first_repetition + args.repetitions)
+            for task in tasks for client in args.clients for arm in args.arms]
+    print(f'{len(jobs)} runs -> {args.out}')
+    if args.dry_run:
+        for client, arm, task, repetition in jobs:
+            print(f'  {client} {arm} {task.name} #{repetition}')
+        return 0
+    args.out.mkdir(parents=True, exist_ok=True)
+    results, lock = [], threading.Lock()
+
+    def record(job):
+        result = run_one(job, args)
+        with lock:
+            results.append(result)
+            with open(args.out / 'results.jsonl', 'a', encoding='utf-8') as stream:
+                stream.write(json.dumps(result) + '\n')
+            print(f'[{len(results)}/{len(jobs)}] {result["client"]} {result["arm"]} {result["task"]} #{result["repetition"]}: '
+                  f'{"PASS" if result["passed"] else "FAIL"} {result["seconds"]}s {result["tool_calls"]} tools', flush=True)
+
+    with ThreadPoolExecutor(max_workers=args.jobs) as pool:
+        list(pool.map(record, jobs))
+    (args.out / 'summary.md').write_text(summarize(results), encoding='utf-8')
+    print(summarize(results))
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

scripts/benchmarks/context.py

@@ -0,0 +1,155 @@
+"""Report the context the installed clients actually load, as the clients report it.
+
+Complements the byte-based prompt budget gate: Claude's own `/context` and `/skill-doctor`
+output and its init event, and Codex's `debug prompt-input`. No model request is made.
+With --check, fail when an installed skill is missing from the client listing or is listed
+without description context (for example because the client's skill budget dropped it).
+"""
+import argparse
+import json
+import os
+from pathlib import Path
+import queue
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import threading
+import tomllib
+
+TIMEOUT = 180
+
+
+def kill_tree(process):
+    if os.name == 'nt':
+        subprocess.run(['taskkill', '/T', '/F', '/PID', str(process.pid)], capture_output=True, timeout=30)
+    else:
+        process.kill()
+    process.wait(timeout=30)
+
+
+def clean_env():
+    return {key: value for key, value in os.environ.items() if not key.startswith(('BUN_INSPECT', 'CLAUDECODE', 'CLAUDE_CODE_ENTRYPOINT'))}
+
+
+def skill_names(root):
+    names = set()
+    for path in root.rglob('SKILL.md') if root.is_dir() else []:
+        match = re.search(r'(?m)^name:\s*"?([^"\r\n]+?)"?\s*$', path.read_text(encoding='utf-8-sig', errors='replace'))
+        if match:
+            names.add(match.group(1))
+    return names
+
+
+def tokens(text):
+    text = text.strip().lstrip('~')
+    if text in ('', '-'):
+        return 0
+    return round(float(text[:-1]) * 1000) if text.endswith('k') else round(float(text))
+
+
+def table_rows(markdown, heading):
+    section = re.search(rf'(?ms)^### {re.escape(heading)}\s*$(.*?)(?=^### |\Z)', markdown)
+    rows = []
+    for line in (section.group(1) if section else '').splitlines():
+        cells = [cell.strip() for cell in line.strip().strip('|').split('|')]
+        if line.startswith('|') and len(cells) >= 2 and not set(cells[0]) <= set('-') and cells[0] not in ('Category', 'Skill', 'Agent Type', 'Type', 'Tool'):
+            rows.append(cells)
+    return rows
+
+
+def claude_print(claude, work, prompt, extra=()):
+    result = subprocess.run([claude, '-p', prompt, '--output-format', 'json', *extra], cwd=work, env=clean_env(),
+                            capture_output=True, text=True, encoding='utf-8', timeout=TIMEOUT)
+    return json.loads(result.stdout).get('result', '')
+
+
+def claude_init(claude, work, extra=()):
+    process = subprocess.Popen([claude, '-p', 'context probe', '--output-format', 'stream-json', '--verbose', *extra], cwd=work,
+                               env=clean_env(), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
+                               text=True, encoding='utf-8')
+    lines = queue.Queue()
+    threading.Thread(target=lambda: [lines.put(line) for line in process.stdout], daemon=True).start()
+    try:
+        while True:
+            event = json.loads(lines.get(timeout=TIMEOUT))
+            if event.get('subtype') == 'init':
+                return event
+    finally:
+        kill_tree(process)
+
+
+def measure_claude(home, work):
+    claude = shutil.which('claude')
+    if not claude:
+        return {'available': False}
+    context = claude_print(claude, work, '/context')
+    doctor = claude_print(claude, work, '/skill-doctor')
+    init = claude_init(claude, work)
+    listed = {row[0]: tokens(row[2]) for row in table_rows(context, 'Skills') if len(row) >= 3}
+    doctor_context = {}
+    for line in doctor.splitlines():
+        cells = line.split()
+        if len(cells) >= 3 and re.fullmatch(r'~?\d+(\.\d+)?k?|-', cells[2]):
+            doctor_context[cells[0]] = tokens(cells[2])
+    installed = skill_names(home / '.claude/skills')
+    return {
+        'available': True,
+        'categories': {row[0]: tokens(row[1]) for row in table_rows(context, 'Estimated usage by category')},
+        'plugins': sorted(plugin.get('name', str(plugin)) if isinstance(plugin, dict) else str(plugin) for plugin in init.get('plugins', [])),
+        'agents': len(init.get('agents', [])),
+        'mcp_servers': len(init.get('mcp_servers', [])),
+        'skills_listed': len(listed),
+        'installed_skills_missing': sorted(installed - set(listed)),
+        'skills_without_context': sorted(name for name, value in {**listed, **doctor_context}.items() if value == 0),
+    }
+
+
+def measure_codex(home, work):
+    codex = shutil.which('codex')
+    if not codex:
+        return {'available': False}
+    result = subprocess.run([codex, 'debug', 'prompt-input', 'context probe'], cwd=work, env=clean_env(),
+                            capture_output=True, text=True, encoding='utf-8', timeout=TIMEOUT)
+    prompt = result.stdout
+    listed = set(re.findall(r'(?m)^- ([^:\n]+?): .+?\(file: ', '\n'.join(
+        part.get('text', '') for item in json.loads(prompt or '[]') for part in item.get('content', []) if isinstance(part, dict))))
+    config_path = home / '.codex/config.toml'
+    config = tomllib.loads(config_path.read_text(encoding='utf-8-sig')) if config_path.is_file() else {}
+    installed = skill_names(home / '.agents/skills')
+    return {
+        'available': True,
+        'prompt_input_tokens_estimate': len(prompt.encode('utf-8')) // 4,
+        'plugins': sorted(name for name, value in config.get('plugins', {}).items() if isinstance(value, dict) and value.get('enabled', True)),
+        'skills_listed': len(listed),
+        'installed_skills_missing': sorted(installed - listed),
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument('--home', type=Path, default=Path.home())
+    parser.add_argument('--json', action='store_true')
+    parser.add_argument('--check', action='store_true')
+    args = parser.parse_args()
+    with tempfile.TemporaryDirectory(prefix='ai-config-context-', ignore_cleanup_errors=True) as directory:
+        subprocess.run(['git', 'init', '-q'], cwd=directory, check=True, timeout=60)
+        report = {'claude': measure_claude(args.home, directory), 'codex': measure_codex(args.home, directory)}
+    if args.json:
+        print(json.dumps(report, indent=2))
+    else:
+        for client, data in report.items():
+            if not data['available']:
+                print(f'{client}: CLI unavailable')
+                continue
+            print(f'{client}: {data["skills_listed"]} skills listed, plugins: {", ".join(data["plugins"]) or "none"}')
+            for key, value in data.items():
+                if key not in ('available', 'plugins', 'skills_listed'):
+                    print(f'  {key}: {value}')
+    problems = [name for data in report.values() if data['available'] for name in data['installed_skills_missing'] + data.get('skills_without_context', [])]
+    return 1 if args.check and problems else 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

scripts/benchmarks/tasks/01-small-config-change/hidden/test_hidden.py

@@ -0,0 +1,16 @@
+import unittest
+from pathlib import Path
+from app import client, config
+
+
+class Hidden(unittest.TestCase):
+    def test_default(self):
+        self.assertEqual(config.DEFAULT_TIMEOUT, 60)
+        self.assertEqual(client.request('x'), 60)
+        self.assertEqual(client.request('x', 5), 5)
+
+    def test_docs(self):
+        for path in ('README.md', 'app/client.py'):
+            text = Path(path).read_text()
+            self.assertIn('60 seconds', text)
+            self.assertNotIn('30 seconds', text)

scripts/benchmarks/tasks/01-small-config-change/repo/README.md

@@ -0,0 +1,3 @@
+# Client
+
+Requests time out after 30 seconds unless a timeout is passed.

scripts/benchmarks/tasks/01-small-config-change/repo/app/client.py

@@ -0,0 +1,6 @@
+from app.config import DEFAULT_TIMEOUT
+
+
+def request(url, timeout=None):
+    """Return the effective timeout in seconds; defaults to DEFAULT_TIMEOUT (30 seconds)."""
+    return timeout if timeout is not None else DEFAULT_TIMEOUT

scripts/benchmarks/tasks/01-small-config-change/repo/app/config.py

@@ -0,0 +1 @@
+DEFAULT_TIMEOUT = 30

scripts/benchmarks/tasks/01-small-config-change/task.json

@@ -0,0 +1,14 @@
+{
+  "prompt": "Change the default request timeout from 30 to 60 seconds.",
+  "workflow": {
+    "kind": "small change without planning ceremony",
+    "forbid_skills": [
+      "superpowers:brainstorming",
+      "superpowers:writing-plans",
+      "superpowers:subagent-driven-development",
+      "superpowers:executing-plans"
+    ],
+    "max_agents": 0,
+    "max_tool_calls": 15
+  }
+}

scripts/benchmarks/tasks/02-bugfix-pagination/hidden/test_hidden.py

@@ -0,0 +1,11 @@
+import unittest
+from app.pagination import paginate
+
+
+class Hidden(unittest.TestCase):
+    def test_last_page(self):
+        self.assertEqual(paginate(list(range(10)), 4, 3), [9])
+        self.assertEqual(paginate(list(range(9)), 3, 3), [6, 7, 8])
+
+    def test_out_of_range(self):
+        self.assertEqual(paginate(list(range(3)), 5, 3), [])

scripts/benchmarks/tasks/02-bugfix-pagination/repo/app/pagination.py

@@ -0,0 +1,5 @@
+def paginate(items, page, size):
+    """Return the 1-based page of items with the given page size."""
+    start = (page - 1) * size
+    end = min(start + size, len(items) - 1)
+    return items[start:end]

scripts/benchmarks/tasks/02-bugfix-pagination/repo/tests/test_pagination.py

@@ -0,0 +1,7 @@
+import unittest
+from app.pagination import paginate
+
+
+class PaginationTests(unittest.TestCase):
+    def test_first_page(self):
+        self.assertEqual(paginate(list(range(10)), 1, 3), [0, 1, 2])

scripts/benchmarks/tasks/02-bugfix-pagination/task.json

@@ -0,0 +1,14 @@
+{
+  "prompt": "Users report that the last item is missing on the final page returned by paginate(). Fix the bug.",
+  "workflow": {
+    "kind": "bugfix with targeted verification",
+    "forbid_skills": [
+      "superpowers:brainstorming",
+      "superpowers:writing-plans",
+      "superpowers:subagent-driven-development",
+      "superpowers:executing-plans"
+    ],
+    "max_agents": 0,
+    "require_command": "test_pagination|pagination"
+  }
+}

scripts/benchmarks/tasks/03-delegation-parsers/hidden/test_hidden.py

@@ -0,0 +1,17 @@
+import unittest
+from app.parsers.csv_parser import parse_csv
+from app.parsers.ini_parser import parse_ini
+from app.parsers.json_parser import parse_json_lines
+
+
+class Hidden(unittest.TestCase):
+    def test_csv(self):
+        self.assertEqual(parse_csv('a, b\n1, "x,y"\n\n2,3\n'), [{'a': '1', 'b': 'x,y'}, {'a': '2', 'b': '3'}])
+
+    def test_json(self):
+        self.assertEqual(parse_json_lines('{"a": 1}\n\n[2]\n'), [{'a': 1}, [2]])
+        with self.assertRaisesRegex(ValueError, '2'):
+            parse_json_lines('{}\n{bad\n')
+
+    def test_ini(self):
+        self.assertEqual(parse_ini('top = 1\n; c\n[s]\n# c\n k = v \n'), {'': {'top': '1'}, 's': {'k': 'v'}})

scripts/benchmarks/tasks/03-delegation-parsers/repo/app/parsers/csv_parser.py

@@ -0,0 +1,6 @@
+def parse_csv(text):
+    """Parse comma-separated text with a header row into a list of dicts.
+
+    Strip whitespace around cells, skip blank lines, and support double-quoted cells containing commas.
+    """
+    raise NotImplementedError

scripts/benchmarks/tasks/03-delegation-parsers/repo/app/parsers/ini_parser.py

@@ -0,0 +1,6 @@
+def parse_ini(text):
+    """Parse INI text into {section: {key: value}}.
+
+    Ignore lines starting with ; or #, strip whitespace, keep keys before the first section under "".
+    """
+    raise NotImplementedError

scripts/benchmarks/tasks/03-delegation-parsers/repo/app/parsers/json_parser.py

@@ -0,0 +1,6 @@
+def parse_json_lines(text):
+    """Parse JSON Lines text into a list of objects.
+
+    Skip blank lines. Raise ValueError naming the 1-based line number for invalid JSON.
+    """
+    raise NotImplementedError

scripts/benchmarks/tasks/03-delegation-parsers/task.json

@@ -0,0 +1,8 @@
+{
+  "prompt": "Implement the three independent parsers in app/parsers (csv_parser, json_parser, ini_parser) according to their docstrings, and add unit tests for each. They share no code.",
+  "workflow": {
+    "kind": "larger change with justified delegation",
+    "min_agents": 1,
+    "max_agents": 3
+  }
+}

scripts/benchmarks/tasks/04-security-sql/hidden/test_hidden.py

@@ -0,0 +1,16 @@
+import sqlite3
+import unittest
+from app.users import find_user
+
+
+class Hidden(unittest.TestCase):
+    def setUp(self):
+        self.conn = sqlite3.connect(':memory:')
+        self.conn.execute('CREATE TABLE users (id INTEGER, name TEXT)')
+        self.conn.executemany('INSERT INTO users VALUES (?, ?)', [(1, "O'Brien"), (2, 'admin')])
+
+    def test_apostrophe(self):
+        self.assertEqual(find_user(self.conn, "O'Brien"), (1, "O'Brien"))
+
+    def test_injection(self):
+        self.assertIsNone(find_user(self.conn, "x' OR '1'='1"))

scripts/benchmarks/tasks/04-security-sql/repo/app/users.py

@@ -0,0 +1,3 @@
+def find_user(conn, name):
+    """Return (id, name) for the user with the given name, or None."""
+    return conn.execute(f"SELECT id, name FROM users WHERE name = '{name}'").fetchone()

scripts/benchmarks/tasks/04-security-sql/task.json

@@ -0,0 +1,6 @@
+{
+  "prompt": "find_user() fails for user names that contain an apostrophe, such as O'Brien. Fix it.",
+  "workflow": {
+    "kind": "security-relevant bugfix"
+  }
+}

scripts/benchmarks/tasks/05-write-tests/hidden/test_hidden.py

@@ -0,0 +1,31 @@
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+MUTANTS = [
+    ('.rstrip("-")', ''),
+    ('.lower()', ''),
+    ('"ascii", "ignore"', '"ascii", "replace"'),
+]
+
+
+def run_tests(root):
+    return subprocess.run([sys.executable, '-m', 'unittest', 'discover', '-s', 'tests', '-t', '.'], cwd=root, capture_output=True, text=True, timeout=120)
+
+
+class Hidden(unittest.TestCase):
+    def test_tests_pass(self):
+        result = run_tests('.')
+        self.assertEqual(result.returncode, 0, result.stderr[-2000:])
+        self.assertRegex(result.stderr, r'Ran ([3-9]|\d\d+) tests')
+
+    def test_tests_catch_mutants(self):
+        source = Path('app/slug.py').read_text()
+        for old, new in MUTANTS:
+            with self.subTest(old), tempfile.TemporaryDirectory() as directory:
+                shutil.copytree('.', directory, dirs_exist_ok=True, ignore=shutil.ignore_patterns('.git', 'bench_hidden'))
+                Path(directory, 'app/slug.py').write_text(source.replace(old, new, 1))
+                self.assertNotEqual(run_tests(directory).returncode, 0, f'tests miss mutant {old!r}')

scripts/benchmarks/tasks/05-write-tests/repo/app/slug.py

@@ -0,0 +1,9 @@
+import re
+import unicodedata
+
+
+def slugify(text, max_length=50):
+    """Lowercase ASCII slug with single hyphens, trimmed to max_length without a trailing hyphen."""
+    text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode()
+    text = re.sub(r"[^a-zA-Z0-9]+", "-", text).strip("-").lower()
+    return text[:max_length].rstrip("-")

scripts/benchmarks/tasks/05-write-tests/task.json

@@ -0,0 +1,6 @@
+{
+  "prompt": "Add unittest-based unit tests for app/slug.py under tests/ that cover its edge cases.",
+  "workflow": {
+    "kind": "test writing"
+  }
+}

scripts/benchmarks/tasks/06-refactor-duplication/hidden/test_hidden.py

@@ -0,0 +1,16 @@
+import unittest
+from pathlib import Path
+from app.report import cost_report, sales_report
+
+ROWS = [(' widgets ', 1234.5), ('gadgets', 99)]
+EXPECTED = 'Widgets                 1,234.50\nGadgets                    99.00\nTotal                   1,333.50'
+
+
+class Hidden(unittest.TestCase):
+    def test_output(self):
+        self.assertEqual(sales_report(ROWS), EXPECTED)
+        self.assertEqual(cost_report(ROWS), EXPECTED)
+        self.assertEqual(sales_report([]), f"{'Total':<20}{0:>12,.2f}")
+
+    def test_deduplicated(self):
+        self.assertEqual(Path('app/report.py').read_text().count('.title()'), 1)

scripts/benchmarks/tasks/06-refactor-duplication/repo/app/report.py

@@ -0,0 +1,20 @@
+def sales_report(rows):
+    lines = []
+    for name, amount in rows:
+        label = name.strip().title()
+        value = f"{amount:,.2f}"
+        lines.append(f"{label:<20}{value:>12}")
+    total = sum(amount for _, amount in rows)
+    lines.append(f"{'Total':<20}{total:>12,.2f}")
+    return "\n".join(lines)
+
+
+def cost_report(rows):
+    lines = []
+    for name, amount in rows:
+        label = name.strip().title()
+        value = f"{amount:,.2f}"
+        lines.append(f"{label:<20}{value:>12}")
+    total = sum(amount for _, amount in rows)
+    lines.append(f"{'Total':<20}{total:>12,.2f}")
+    return "\n".join(lines)

scripts/benchmarks/tasks/06-refactor-duplication/task.json

@@ -0,0 +1,6 @@
+{
+  "prompt": "Remove the duplicated formatting logic in app/report.py without changing its output.",
+  "workflow": {
+    "kind": "behavior-preserving refactor"
+  }
+}

scripts/benchmarks/tasks/07-preserve-unknown-settings/hidden/test_hidden.py

@@ -0,0 +1,20 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from app.settings_store import save_settings
+
+
+class Hidden(unittest.TestCase):
+    def test_unknown_keys_survive(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'settings.json'
+            path.write_text(json.dumps({'theme': 'dark', 'plugins': {'a': True}, 'x': 1}))
+            save_settings(path, {'language': 'de'})
+            self.assertEqual(json.loads(path.read_text()), {'theme': 'dark', 'plugins': {'a': True}, 'x': 1, 'language': 'de'})
+
+    def test_new_file(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'settings.json'
+            save_settings(path, {'theme': 'light'})
+            self.assertEqual(json.loads(path.read_text()), {'theme': 'light'})

scripts/benchmarks/tasks/07-preserve-unknown-settings/repo/app/settings_store.py

@@ -0,0 +1,15 @@
+import json
+
+KNOWN = ("theme", "language")
+
+
+def save_settings(path, updates):
+    """Apply updates to the JSON settings file at path."""
+    try:
+        with open(path, encoding="utf-8") as stream:
+            current = json.load(stream)
+    except FileNotFoundError:
+        current = {}
+    merged = {key: updates.get(key, current.get(key)) for key in KNOWN if key in updates or key in current}
+    with open(path, "w", encoding="utf-8") as stream:
+        json.dump(merged, stream, indent=2)

scripts/benchmarks/tasks/07-preserve-unknown-settings/task.json

@@ -0,0 +1,6 @@
+{
+  "prompt": "save_settings() loses keys it does not know about when it updates the settings file. Fix it so unknown keys survive.",
+  "workflow": {
+    "kind": "data-preserving bugfix"
+  }
+}

scripts/benchmarks/tasks/08-api-rename-with-docs/hidden/test_hidden.py

@@ -0,0 +1,15 @@
+import inspect
+import unittest
+from pathlib import Path
+from app import api, cli
+
+
+class Hidden(unittest.TestCase):
+    def test_signature(self):
+        self.assertEqual(list(inspect.signature(api.fetch).parameters), ['url', 'max_retries'])
+        self.assertEqual(cli.main('u'), 2)
+
+    def test_docs(self):
+        text = Path('README.md').read_text()
+        self.assertIn('max_retries', text)
+        self.assertNotRegex(text, r'(?<![\w])retries(?=[=`])')

scripts/benchmarks/tasks/08-api-rename-with-docs/repo/README.md

@@ -0,0 +1,3 @@
+# API
+
+`fetch(url, retries=3)` retries a request up to `retries` times.

scripts/benchmarks/tasks/08-api-rename-with-docs/repo/app/api.py

@@ -0,0 +1,3 @@
+def fetch(url, retries=3):
+    """Fetch url, retrying up to retries times. Returns the attempts used."""
+    return min(retries, 3) if url else 0

scripts/benchmarks/tasks/08-api-rename-with-docs/repo/app/cli.py

@@ -0,0 +1,5 @@
+from app.api import fetch
+
+
+def main(url):
+    return fetch(url, retries=2)

scripts/benchmarks/tasks/08-api-rename-with-docs/task.json

@@ -0,0 +1,6 @@
+{
+  "prompt": "Rename the retries parameter of app.api.fetch to max_retries.",
+  "workflow": {
+    "kind": "API change with directly affected documentation"
+  }
+}

scripts/benchmarks/tasks/09-path-traversal/hidden/test_hidden.py

@@ -0,0 +1,26 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from app.files import read_upload
+
+
+class Hidden(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        root = Path(self.directory.name)
+        (root / 'uploads/sub').mkdir(parents=True)
+        (root / 'uploads/sub/a.txt').write_bytes(b'ok')
+        (root / 'secret.txt').write_bytes(b'secret')
+        self.base = str(root / 'uploads')
+
+    def tearDown(self):
+        self.directory.cleanup()
+
+    def test_nested(self):
+        self.assertEqual(read_upload(self.base, 'sub/a.txt'), b'ok')
+
+    def test_traversal(self):
+        for name in ('../secret.txt', 'sub/../../secret.txt', os.path.join(os.path.dirname(self.base), 'secret.txt')):
+            with self.subTest(name), self.assertRaises((ValueError, PermissionError, FileNotFoundError)):
+                read_upload(self.base, name)

scripts/benchmarks/tasks/09-path-traversal/repo/app/files.py

@@ -0,0 +1,7 @@
+import os
+
+
+def read_upload(base, name):
+    """Return the bytes of the uploaded file name below the base directory."""
+    with open(os.path.join(base, name), "rb") as stream:
+        return stream.read()

scripts/benchmarks/tasks/09-path-traversal/task.json

@@ -0,0 +1,6 @@
+{
+  "prompt": "Make read_upload() safe against path traversal. Legitimate names in nested folders must keep working.",
+  "workflow": {
+    "kind": "security hardening"
+  }
+}

scripts/benchmarks/tasks/10-performance-dedupe/hidden/test_hidden.py

@@ -0,0 +1,16 @@
+import time
+import unittest
+from app.dedupe import dedupe
+
+
+class Hidden(unittest.TestCase):
+    def test_behavior(self):
+        self.assertEqual(dedupe([3, 1, 3, 2, 1]), [3, 1, 2])
+        self.assertEqual(dedupe([]), [])
+        self.assertEqual(dedupe(['a', 'b', 'a']), ['a', 'b'])
+
+    def test_fast(self):
+        items = list(range(100000)) * 2
+        start = time.perf_counter()
+        self.assertEqual(len(dedupe(items)), 100000)
+        self.assertLess(time.perf_counter() - start, 1.0)

scripts/benchmarks/tasks/10-performance-dedupe/repo/app/dedupe.py

@@ -0,0 +1,7 @@
+def dedupe(items):
+    """Return items without duplicates, keeping the first occurrence of each."""
+    result = []
+    for item in items:
+        if item not in result:
+            result.append(item)
+    return result

scripts/benchmarks/tasks/10-performance-dedupe/task.json

@@ -0,0 +1,6 @@
+{
+  "prompt": "dedupe() is far too slow for 100,000 items. Make it fast while keeping its behavior and the order of first occurrences.",
+  "workflow": {
+    "kind": "performance fix"
+  }
+}

scripts/commands/benchmark.ps1

@@ -0,0 +1,4 @@
+$ErrorActionPreference = 'Stop'
+$root = Split-Path (Split-Path $PSScriptRoot -Parent) -Parent
+& python (Join-Path $root 'scripts/benchmarks/ab.py') @args
+exit $LASTEXITCODE

scripts/commands/benchmark.sh

@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -euo pipefail
+root=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/../.." && pwd)
+exec python3 "$root/scripts/benchmarks/ab.py" "$@"

scripts/commands/context-report.ps1

@@ -0,0 +1,4 @@
+$ErrorActionPreference = 'Stop'
+$root = Split-Path (Split-Path $PSScriptRoot -Parent) -Parent
+& python (Join-Path $root 'scripts/benchmarks/context.py') @args
+exit $LASTEXITCODE

scripts/commands/context-report.sh

@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -euo pipefail
+root=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/../.." && pwd)
+exec python3 "$root/scripts/benchmarks/context.py" "$@"

scripts/lib/manifest.ps1

@@ -43,9 +43,9 @@ function Get-Sha256HashOfBytes {
     finally { $sha256.Dispose() }
 }
 
-function Write-ManagedSkillFile {
-    param([Parameter(Mandatory=$true)][string]$Path,[Parameter(Mandatory=$true)][byte[]]$Bytes)
-    $temporary = Join-Path (Split-Path $Path -Parent) ('.ai-config-skill-' + [Guid]::NewGuid().ToString('N') + '.tmp')
+function Write-ManagedFileAtomically {
+    param([Parameter(Mandatory=$true)][string]$Path,[Parameter(Mandatory=$true)][AllowEmptyCollection()][byte[]]$Bytes)
+    $temporary = Join-Path (Split-Path $Path -Parent) ('.ai-config-write-' + [Guid]::NewGuid().ToString('N') + '.tmp')
     $backup = "$temporary.bak"
     try {
         [System.IO.File]::WriteAllBytes($temporary, $Bytes)
@@ -83,7 +83,27 @@ function Write-ManagedManifest {
     # without BOM and with LF line endings so either implementation can read the file
     # regardless of which one wrote it.
     $content = ($lines -join "`n") + "`n"
-    [System.IO.File]::WriteAllText($ManifestPath, $content, (New-Object System.Text.UTF8Encoding($false)))
+    Write-ManagedFileAtomically -Path $ManifestPath -Bytes ((New-Object System.Text.UTF8Encoding($false)).GetBytes($content))
+}
+
+function Enter-ManagedLock {
+    param([Parameter(Mandatory=$true)][string]$Destination)
+    New-Item $Destination -ItemType Directory -Force | Out-Null
+    $lockPath = Join-Path $Destination '.ai-config.lock'
+    $owner = "powershell $([Environment]::MachineName) $PID"
+    foreach ($attempt in 1, 2) {
+        try {
+            $stream = [System.IO.File]::Open($lockPath, [System.IO.FileMode]::CreateNew, [System.IO.FileAccess]::Write, [System.IO.FileShare]::None)
+            try { $bytes = [System.Text.Encoding]::UTF8.GetBytes("$owner`n"); $stream.Write($bytes, 0, $bytes.Length) } finally { $stream.Dispose() }
+            return $lockPath
+        } catch [System.IO.IOException] {
+            $holder = @(([System.IO.File]::ReadAllText($lockPath)).Trim() -split ' ')
+            # ponytail: only same-shell, same-host owners are checked for staleness; other stale locks need manual removal.
+            $stale = $attempt -eq 1 -and $holder.Count -eq 3 -and $holder[0] -eq 'powershell' -and $holder[1] -eq [Environment]::MachineName -and -not (Get-Process -Id ([int]$holder[2]) -ErrorAction SilentlyContinue)
+            if (-not $stale) { throw "Another installation or update holds $lockPath; wait for it to finish, or remove the file if no update is running." }
+            [System.IO.File]::Delete($lockPath)
+        }
+    }
 }
 
 function Invoke-InstallOptions {
@@ -107,6 +127,30 @@ function Invoke-InstallOptions {
 }
 
 function Sync-ManagedDestination {
+    <#
+    .SYNOPSIS
+        Runs Invoke-ManagedSync while holding the destination lock, so parallel
+        installs or updates cannot interleave writes to the same destination.
+    #>
+    param(
+        [Parameter(Mandatory=$true)][string]$Source,
+        [Parameter(Mandatory=$true)][string]$Destination,
+        [Parameter(Mandatory=$true)][string]$Stamp,
+        [string]$AiConfigRoot,
+        [string]$ShellCommand,
+        [string]$PowerShellCommand,
+        [string]$ClaudeConcurrency = '5',
+        [bool]$FlashbangEnabled = $true,
+        [bool]$StatusLineEnabled = $true,
+        [switch]$DryRun,
+        [switch]$Summary
+    )
+    if ($DryRun) { return Invoke-ManagedSync @PSBoundParameters }
+    $lockPath = Enter-ManagedLock $Destination
+    try { Invoke-ManagedSync @PSBoundParameters } finally { [System.IO.File]::Delete($lockPath) }
+}
+
+function Invoke-ManagedSync {
     <#
     .SYNOPSIS
         Installs one generated source tree into a destination directory, tracking
@@ -190,11 +234,7 @@ function Sync-ManagedDestination {
             if (-not $wasManaged) { Write-Output "WARN $target existed before this installation but was not tracked by a previous run; it was backed up before being overwritten."; $tally.Warned++ }
         }
         New-Item (Split-Path $target -Parent) -ItemType Directory -Force | Out-Null
-        if ($relative -eq 'SKILL.md' -or $relative.EndsWith('/SKILL.md', [StringComparison]::Ordinal)) {
-            Write-ManagedSkillFile -Path $target -Bytes $newBytes
-        } else {
-            [System.IO.File]::WriteAllBytes($target, $newBytes)
-        }
+        Write-ManagedFileAtomically -Path $target -Bytes $newBytes
         if (-not $Summary) { Write-Output "$action $target" }
         if ($action -eq 'CREATE') { $tally.Created++ } else { $tally.Updated++ }
     }

scripts/lib/manifest.sh

@@ -66,22 +66,62 @@ read_managed_manifest() {
   done < "$path"
 }
 
+# write_managed_file <target> <content-file> [source-file]
+# Replace target atomically: write a sibling temporary file, then rename it over the target.
+write_managed_file() {
+  local target=$1 content_file=$2 source_file=${3:-} temporary mode
+  temporary=$(mktemp "$(dirname -- "$target")/.ai-config-write.XXXXXX") || return 1
+  mode=$(printf '%o' $((0666 & ~$(umask))))
+  [ -z "$source_file" ] || [ ! -x "$source_file" ] || mode=$(printf '%o' $((0777 & ~$(umask))))
+  { cat -- "$content_file" > "$temporary" && chmod "$mode" "$temporary" && mv -f -- "$temporary" "$target"; } || { rm -f -- "$temporary"; return 1; }
+}
+
 # write_managed_manifest <manifest-file> <assoc-array-name>
 write_managed_manifest() {
-  local path=$1
+  local path=$1 content
   local -n in_ref=$2
   mkdir -p "$(dirname -- "$path")"
+  content=$(mktemp) || return 1
   {
     printf 'path\tsha256\n'
     while IFS= read -r rel; do
       [ -n "$rel" ] || continue
       printf '%s\t%s\n' "$rel" "${in_ref[$rel]}"
     done < <(printf '%s\n' "${!in_ref[@]}" | sort)
-  } > "$path"
+  } > "$content" && write_managed_file "$path" "$content"
+  local status=$?
+  rm -f -- "$content"
+  return "$status"
 }
 
-# sync_managed_destination <source-dir> <destination-dir> <stamp> <ai-config-root> <shell-command> <powershell-command> <dry-run: true|false> [summary: true|false] [flashbang] [claude-concurrency] [statusline]
+# acquire_managed_lock <destination>: create <destination>/.ai-config.lock exclusively.
+acquire_managed_lock() {
+  local lock="$1/.ai-config.lock" owner holder_shell holder_host holder_pid
+  mkdir -p -- "$1"
+  owner="bash $(uname -n) $$"
+  if (set -o noclobber; printf '%s\n' "$owner" > "$lock") 2>/dev/null; then return 0; fi
+  read -r holder_shell holder_host holder_pid < "$lock" 2>/dev/null || true
+  # ponytail: only same-shell, same-host owners are checked for staleness; other stale locks need manual removal.
+  if [ "$holder_shell" = bash ] && [ "$holder_host" = "$(uname -n)" ] && [[ "$holder_pid" =~ ^[0-9]+$ ]] && ! kill -0 "$holder_pid" 2>/dev/null; then
+    rm -f -- "$lock"
+    if (set -o noclobber; printf '%s\n' "$owner" > "$lock") 2>/dev/null; then return 0; fi
+  fi
+  printf 'Another installation or update holds %s; wait for it to finish, or remove the file if no update is running.\n' "$lock" >&2
+  return 1
+}
+
+# sync_managed_destination: same arguments as run_managed_sync; holds the destination lock unless dry-running.
 sync_managed_destination() {
+  [ "$7" != true ] || { run_managed_sync "$@"; return; }
+  acquire_managed_lock "$2" || return 1
+  local status=0
+  run_managed_sync "$@" || status=$?
+  rm -f -- "$2/.ai-config.lock"
+  return "$status"
+}
+
+# run_managed_sync <source-dir> <destination-dir> <stamp> <ai-config-root> <shell-command> <powershell-command> <dry-run: true|false> [summary: true|false] [flashbang] [claude-concurrency] [statusline]
+run_managed_sync() {
   local source=$1 destination=$2 stamp=$3 ai_config_root=$4 shell_command=$5 powershell_command=$6 dry_run=$7 summary=${8:-false} flashbang_enabled=${9:-true} claude_concurrency=${10:-5} statusline_enabled=${11:-true}
   [ "$summary" = true ] || printf 'SOURCE %s -> %s\n' "$source" "$destination"
   local manifest
@@ -176,19 +216,13 @@ sync_managed_destination() {
       fi
     fi
     mkdir -p "$(dirname -- "$target")"
-    if [[ "$relative" = SKILL.md || "$relative" = */SKILL.md ]]; then
-      local temporary
-      temporary=$(mktemp "$(dirname -- "$target")/.ai-config-skill.XXXXXX") || return 1
-      if [ "$needs_sub" = true ]; then
-        printf '%s' "$content" > "$temporary" || { rm -f -- "$temporary"; return 1; }
-      else
-        cp -- "$source_file" "$temporary" || { rm -f -- "$temporary"; return 1; }
-      fi
-      mv -f -- "$temporary" "$target" || { rm -f -- "$temporary"; return 1; }
-    elif [ "$needs_sub" = true ]; then
-      printf '%s' "$content" > "$target"
+    if [ "$needs_sub" = true ]; then
+      local rendered
+      rendered=$(mktemp) || return 1
+      printf '%s' "$content" > "$rendered" && write_managed_file "$target" "$rendered" "$source_file" || { rm -f -- "$rendered"; return 1; }
+      rm -f -- "$rendered"
     else
-      cp -- "$source_file" "$target"
+      write_managed_file "$target" "$source_file" "$source_file" || return 1
     fi
     [ "$summary" = true ] || printf '%s %s\n' "$action" "$target"
     [ "$action" = CREATE ] && created=$((created + 1)) || updated=$((updated + 1))
@@ -219,7 +253,7 @@ sync_managed_destination() {
     fi
   done
 
-  if [ "$dry_run" != true ]; then write_managed_manifest "$manifest" new_manifest; fi
+  if [ "$dry_run" != true ]; then write_managed_manifest "$manifest" new_manifest || return 1; fi
   if [ "$summary" = true ]; then
     printf 'SYNC %s: %s created, %s updated, %s unchanged, %s removed, %s warnings\n' "$destination" "$created" "$updated" "$unchanged" "$removed" "$warned"
   fi

scripts/tests/test-managed-install.ps1

@@ -111,7 +111,7 @@ try {
     $installedSkill = Join-Path $testHome '.agents/skills/example/SKILL.md'
     if ((Get-Content -LiteralPath $installedSkill -Raw) -ne 'updated managed skill') { throw 'Managed skill update did not replace the file.' }
     if ((Get-Content -LiteralPath (Join-Path $skillBackupRoot 'skill-update/example/SKILL.md') -Raw) -ne 'managed skill') { throw 'New skill backup must be outside the skill tree.' }
-    if (@(Get-ChildItem -LiteralPath (Split-Path $installedSkill -Parent) -Filter '.ai-config-skill-*' -File).Count) { throw 'Managed skill update left a temporary file.' }
+    if (@(Get-ChildItem -LiteralPath (Split-Path $installedSkill -Parent) -Filter '.ai-config-write-*' -File).Count) { throw 'Managed skill update left a temporary file.' }
     Remove-Item -LiteralPath (Join-Path $package 'skills/example/SKILL.md') -Force
     $null = Sync-ManagedDestination -Source $targets[0].Source -Destination $targets[0].Destination -Stamp 'skill-remove'
     if (Test-Path -LiteralPath $installedSkill) { throw 'Deselected managed skill must be removed.' }
@@ -119,6 +119,21 @@ try {
     Set-Content (Join-Path $source 'concurrency.txt') '__CLAUDE_CODE_MAX_TOOL_USE_CONCURRENCY__' -NoNewline
     $null = Sync-ManagedDestination -Source $source -Destination $destination -Stamp 'concurrency' -ClaudeConcurrency '7'
     if ((Get-Content (Join-Path $destination 'concurrency.txt') -Raw) -ne '7') { throw 'Standalone concurrency placeholder must be resolved.' }
+    if (@(Get-ChildItem -LiteralPath $destination -Filter '.ai-config-write-*' -File -Recurse).Count) { throw 'Atomic writes left a temporary file.' }
+    if (Test-Path -LiteralPath (Join-Path $destination '.ai-config.lock')) { throw 'Sync must release its destination lock.' }
+
+    $lockPath = Join-Path $destination '.ai-config.lock'
+    [System.IO.File]::WriteAllText($lockPath, "bash other-host 1`n")
+    $manifestBefore = [System.IO.File]::ReadAllText((Get-ManagedManifestPath $destination))
+    $blocked = $false
+    try { $null = Sync-ManagedDestination -Source $source -Destination $destination -Stamp 'locked' } catch { $blocked = $_.Exception.Message -like 'Another installation or update holds*' }
+    if (-not $blocked) { throw 'A held destination lock must block a parallel sync.' }
+    if ([System.IO.File]::ReadAllText((Get-ManagedManifestPath $destination)) -ne $manifestBefore) { throw 'A blocked sync must not write the manifest.' }
+    $exited = Start-Process -FilePath (Get-Process -Id $PID).Path -ArgumentList '-NoProfile', '-Command', 'exit 0' -PassThru -WindowStyle Hidden
+    $exited.WaitForExit()
+    [System.IO.File]::WriteAllText($lockPath, "powershell $([Environment]::MachineName) $($exited.Id)`n")
+    $null = Sync-ManagedDestination -Source $source -Destination $destination -Stamp 'stale-lock'
+    if (Test-Path -LiteralPath $lockPath) { throw 'A stale lock from an exited process must be replaced and released.' }
 
     if ($Summary) { Write-Output 'Tests: PASS | managed install' } else { Write-Output 'PASS managed manifest: idempotent, stale removal, modified-file protection, foreign files preserved, dry run side-effect free, skill migration' }
 } finally {

scripts/tests/test-managed-install.sh

@@ -128,12 +128,23 @@ for target_pair in "${targets[@]}"; do
 done
 [ "$(cat "$test_home/.agents/skills/example/SKILL.md")" = 'updated managed skill' ] || { printf 'Managed skill update did not replace the file.\n' >&2; exit 1; }
 [ "$(cat "$test_home/.agents/.ai-config-skill-backups/skill-update/example/SKILL.md")" = 'managed skill' ] || { printf 'New skill backup must be outside the skill tree.\n' >&2; exit 1; }
-[ -z "$(find "$test_home/.agents/skills/example" -maxdepth 1 -name '.ai-config-skill.*' -print -quit)" ] || { printf 'Managed skill update left a temporary file.\n' >&2; exit 1; }
+[ -z "$(find "$test_home/.agents/skills/example" -maxdepth 1 -name '.ai-config-write.*' -print -quit)" ] || { printf 'Managed skill update left a temporary file.\n' >&2; exit 1; }
 rm -f -- "$package/skills/example/SKILL.md"
 sync_managed_destination "${targets[0]%%|*}" "${targets[0]#*|}" skill-remove "" "" "" false > /dev/null
 [ ! -f "$test_home/.agents/skills/example/SKILL.md" ] && [ "$(cat "$test_home/.agents/.ai-config-skill-backups/skill-remove/example/SKILL.md")" = 'updated managed skill' ] || { printf 'Deselected skill must be removed and backed up outside the skill tree.\n' >&2; exit 1; }
 printf '__CLAUDE_CODE_MAX_TOOL_USE_CONCURRENCY__' > "$source_dir/concurrency.txt"
 sync_managed_destination "$source_dir" "$destination" concurrency "" "" "" false false true 7 > /dev/null
 [ "$(cat "$destination/concurrency.txt")" = 7 ] || { printf 'Standalone concurrency placeholder must be resolved.\n' >&2; exit 1; }
+[ -z "$(find "$destination" -name '.ai-config-write.*' -print -quit)" ] || { printf 'Atomic writes left a temporary file.\n' >&2; exit 1; }
+[ ! -e "$destination/.ai-config.lock" ] || { printf 'Sync must release its destination lock.\n' >&2; exit 1; }
+printf 'powershell other-host 1\n' > "$destination/.ai-config.lock"
+manifest_before=$(cat "$destination/.ai-config-manifest.tsv")
+if sync_managed_destination "$source_dir" "$destination" locked "" "" "" false > /dev/null 2>&1; then printf 'A held destination lock must block a parallel sync.\n' >&2; exit 1; fi
+[ "$(cat "$destination/.ai-config-manifest.tsv")" = "$manifest_before" ] || { printf 'A blocked sync must not write the manifest.\n' >&2; exit 1; }
+bash -c 'exit 0' & exited_pid=$!
+wait "$exited_pid"
+printf 'bash %s %s\n' "$(uname -n)" "$exited_pid" > "$destination/.ai-config.lock"
+sync_managed_destination "$source_dir" "$destination" stale-lock "" "" "" false > /dev/null
+[ ! -e "$destination/.ai-config.lock" ] || { printf 'A stale lock from an exited process must be replaced and released.\n' >&2; exit 1; }
 
 if [ "$summary" = true ]; then printf 'Tests: PASS | managed install\n'; else printf 'PASS managed manifest: idempotent, stale removal, modified-file protection, foreign files preserved, dry run side-effect free, skill migration\n'; fi