"""Explicit pool-owned Codex settings, applied only after an operator opts in.""" import argparse import json import os from pathlib import Path import re import subprocess import sys import tempfile from codex_python import require_python311 require_python311(__file__) from codex_config_edit import limits, tomllib, updated_config, validate_overrides POLICY_FILE = "codex-settings-policy.json" def positive_integer(value): if isinstance(value, bool) or not re.fullmatch(r"[0-9]+", str(value)): raise ValueError("max-subagents must be a positive integer") parsed = int(value) if parsed < 1: raise ValueError("max-subagents must be a positive integer") return parsed def policy_value(root): path = root / POLICY_FILE if not path.exists(): return None policy = json.loads(path.read_text()) if policy.get("version") != 1: raise ValueError("unsupported Codex settings policy version") return positive_integer(policy["max_subagents"]) def target_paths(root, include_global): manifest = json.loads((root / "accounts.json").read_text()) accounts = manifest.get("accounts") if not isinstance(accounts, list): raise ValueError("manifest accounts must be a list") result = [] for account in accounts: account_id = account.get("id", "") if isinstance(account, dict) else "" if not re.fullmatch(r"acct-\d{2}", account_id): raise ValueError("manifest has an invalid account id") result.append(root / account_id / "config.toml") if include_global: result.append(Path.home() / ".codex" / "config.toml") return list(dict.fromkeys(result)) def atomic_write(path, text): target = path.resolve() target.parent.mkdir(parents=True, exist_ok=True) mode = target.stat().st_mode & 0o777 if target.exists() else 0o600 descriptor, name = tempfile.mkstemp(prefix=".codex-settings-", dir=target.parent) try: with os.fdopen(descriptor, "w") as handle: os.fchmod(handle.fileno(), mode) handle.write(text) os.replace(name, target) finally: if os.path.exists(name): os.unlink(name) def configure(root, paths, value): pending, seen = [], set() for path in paths: if path.resolve() in seen: continue seen.add(path.resolve()) previous = path.read_text() if path.exists() else "" source = previous if not path.exists() and path != Path.home() / ".codex/config.toml": global_config = Path.home() / ".codex/config.toml" source = global_config.read_text() if global_config.exists() else "" pending.append((path, previous, updated_config(source, value))) for path, previous, updated in pending: if (path.read_text() if path.exists() else "") != previous: raise ValueError("configuration changed during validation; retry configure") if updated != previous: atomic_write(path, updated) atomic_write(root / POLICY_FILE, json.dumps({"version": 1, "max_subagents": value}) + "\n") def readback(paths, value): result = {} for path in paths: document = tomllib.loads(path.read_text()) if path.exists() else {} current = limits(document) matches = (all(current[key] == value for key in ("configured", "feature_override")) if value is not None else True) try: validate_overrides(document, value) except ValueError: matches = False result[path] = matches return result def plugin_version(): repo = Path(__file__).resolve().parents[1] if (repo / ".git").exists(): try: result = subprocess.run(["git", "describe", "--tags", "--exact-match"], cwd=repo, capture_output=True, text=True, check=False, timeout=5) tag = result.stdout.strip() if result.returncode == 0 and re.fullmatch(r"v\d+\.\d+\.\d+", tag): return tag[1:] except (OSError, subprocess.TimeoutExpired): pass return str(json.loads((repo / "package.json").read_text())["version"]) def report_for(root, paths, value, include_global): matches = readback(paths, value) global_path = Path.home() / ".codex/config.toml" global_configured = matches.pop(global_path, None) if include_global else None configured, total = sum(matches.values()), len(matches) policy = policy_value(root) mismatches = total - configured + int(global_configured is False) return {"provider": "codex", "plugin_version": plugin_version(), "max_subagents": value, "policy_max_subagents": policy, "accounts_total": total, "accounts_configured": configured, "global_configured": global_configured, "mismatch_count": mismatches, "verified": mismatches == 0 and policy == value} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("mode", choices=("configure", "seed", "policy-value")) parser.add_argument("root", type=Path) parser.add_argument("--max-subagents", type=positive_integer) parser.add_argument("--include-global", action="store_true") parser.add_argument("--json", action="store_true") parser.add_argument("--check", action="store_true") parser.add_argument("--account-dir", type=Path) args = parser.parse_args() value = args.max_subagents if args.max_subagents is not None else policy_value(args.root) if args.mode == "policy-value": if value is not None: print(value) return if args.mode == "seed": if value is not None: path = args.account_dir / "config.toml" source = path.read_text() if path.exists() else "" updated = updated_config(source, value) if source != updated: atomic_write(path, updated) return paths = target_paths(args.root, args.include_global) if args.max_subagents is not None and not args.check: configure(args.root, paths, value) report = report_for(args.root, paths, value, args.include_global) print(json.dumps(report) if args.json else f"Codex subagent policy: {value if value is not None else 'unmanaged'}; checked {len(paths)} configs") if not report["verified"]: sys.exit(1) if __name__ == "__main__": try: main() except (ValueError, OSError, KeyError, TypeError) as error: print(f"codex-accounts configure: {error}", file=sys.stderr) sys.exit(1)