Setting the file. One moment. 11.10
OpenAI
scripts/stage_install.py
Python·184 lines·7 KB
import
datetime, timezone
14from pathlib import Path
15
16
17NAME_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
18
19
20def parse_name(skill_file: Path) -> str:
21 lines = skill_file.read_text(encoding="utf-8", errors="strict").splitlines()
22 if not lines or lines[0].strip() != "---":
23 raise ValueError(f"{skill_file}: missing opening frontmatter delimiter")
24 end = next((i for i in range(1, len(lines)) if lines[i].strip() == "---"), None)
25 if end is None:
26 raise ValueError(f"{skill_file}: missing closing frontmatter delimiter")
27 for line in lines[1:end]:
28 match = re.match(r"^name:\s*([^#]+?)\s*$", line)
29 if match:
30 name = match.group(1).strip().strip('"\'')
31 if len(name) > 63 or not NAME_RE.fullmatch(name):
32 raise ValueError(f"{skill_file}: invalid skill name {name!r}")
33 return name
34 raise ValueError(f"{skill_file}: missing name")
35
36
37def collect_files(source: Path) -> list[Path]:
38 files: list[Path] = []
39 for current, dirs, filenames in os.walk(source, followlinks=False):
40 current_path = Path(current)
41 for dirname in dirs:
42 path = current_path / dirname
43 if path.is_symlink():
44 raise ValueError(f"symlinked directories are not allowed: {path}")
45 for filename in filenames:
46 path = current_path / filename
47 if path.is_symlink():
48 raise ValueError(f"symlinked files are not allowed: {path}")
49 if not path.is_file():
50 raise ValueError(f"unsupported filesystem entry: {path}")
51 files.append(path)
52 return sorted(files, key=lambda path: str(path.relative_to(source)))
53
54
55def sha256_file(path: Path) -> str:
56 digest = hashlib.sha256()
57 with path.open("rb") as handle:
58 for chunk in iter(lambda: handle.read(1024 * 1024), b""):
59 digest.update(chunk)
60 return digest.hexdigest()
61
62
63def source_record(source: Path, dest: Path) -> dict[str, object]:
64 if source.is_symlink() or not source.is_dir():
65 raise ValueError(f"source is not a regular directory: {source}")
66 skill_file = source / "SKILL.md"
67 if not skill_file.is_file():
68 raise ValueError(f"source has no SKILL.md: {source}")
69 name = parse_name(skill_file)
70 files = collect_files(source)
71 file_records = [
72 {
73 "path": str(path.relative_to(source)),
74 "sha256": sha256_file(path),
75 "bytes": path.stat().st_size,
76 }
77 for path in files
78 ]
79 aggregate = hashlib.sha256()
80 for item in file_records:
81 aggregate.update(str(item["path"]).encode("utf-8"))
82 aggregate.update(str(item["sha256"]).encode("ascii"))
83 return {
84 "name": name,
85 "source": str(source),
86 "target": str(dest / name),
87 "content_sha256": aggregate.hexdigest(),
88 "file_count": len(file_records),
89 "files": file_records,
90 }
91
92
93def atomic_write_json(path: Path, payload: dict[str, object]) -> None:
94 path.parent.mkdir(parents=True, exist_ok=True)
95 with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle:
96 json.dump(payload, handle, ensure_ascii=False, indent=2)
97 handle.write("\n")
98 temp_name = handle.name
99 os.replace(temp_name, path)
100
101
102def apply_install(records: list[dict[str, object]], dest: Path) -> list[str]:
103 dest.mkdir(parents=True, exist_ok=True)
104 staging = Path(tempfile.mkdtemp(prefix=".agent-skill-stack-", dir=dest))
105 created: list[str] = []
106 try:
107 for record in records:
108 source = Path(str(record["source"]))
109 staged = staging / str(record["name"])
110 shutil.copytree(source, staged, symlinks=False)
111 if parse_name(staged / "SKILL.md") != record["name"]:
112 raise RuntimeError(f"staged validation failed for {record['name']}")
113 for record in records:
114 staged = staging / str(record["name"])
115 target = Path(str(record["target"]))
116 os.replace(staged, target)
117 created.append(str(target))
118 return created
119 finally:
120 shutil.rmtree(staging, ignore_errors=True)
121
122
123def main() -> int:
124 parser = argparse.ArgumentParser(description=__doc__)
125 parser.add_argument("--source", action="append", required=True, help="Audited local skill directory; repeatable")
126 parser.add_argument("--dest", required=True, help="Destination skill root")
127 parser.add_argument("--manifest", required=True, help="Path for the lock/preview manifest")
128 parser.add_argument("--apply", action="store_true", help="Copy after validation; default is dry-run")
129 parser.add_argument(
130 "--record-existing",
131 action="store_true",
132 help="Record a lock manifest when each source is already its exact destination",
133 )
134 args = parser.parse_args()
135
136 if args.apply and args.record_existing:
137 raise SystemExit("--apply and --record-existing are mutually exclusive")
138
139 dest = Path(os.path.expandvars(os.path.expanduser(args.dest))).resolve()
140 manifest = Path(os.path.expandvars(os.path.expanduser(args.manifest))).resolve()
141 sources = [Path(os.path.expandvars(os.path.expanduser(raw))).resolve() for raw in args.source]
142
143 records = [source_record(source, dest) for source in sources]
144 names = [str(record["name"]) for record in records]
145 if len(names) != len(set(names)):
146 raise SystemExit("duplicate skill names in selected sources")
147
148 existing = [str(record["target"]) for record in records if Path(str(record["target"])).exists()]
149 if args.record_existing:
150 mismatched = [
151 str(record["name"])
152 for record in records
153 if Path(str(record["source"])).resolve() != Path(str(record["target"])).resolve()
154 ]
155 if mismatched:
156 raise SystemExit("--record-existing requires source to equal target for: " + ", ".join(mismatched))
157 elif existing:
158 raise SystemExit("refusing to overwrite existing destinations: " + ", ".join(existing))
159
160 payload: dict[str, object] = {
161 "schema": 1,
162 "generated_at": datetime.now(timezone.utc).isoformat(),
163 "mode": "record-existing" if args.record_existing else ("apply" if args.apply else "dry-run"),
164 "destination": str(dest),
165 "skills": records,
166 "created": [],
167 "notice": "Sources must be downloaded and audited before using this installer. Existing targets are never overwritten.",
168 }
169
170 if args.record_existing:
171 payload["status"] = "recorded"
172 elif args.apply:
173 payload["created"] = apply_install(records, dest)
174 payload["status"] = "installed"
175 else:
176 payload["status"] = "planned"
177
178 atomic_write_json(manifest, payload)
179 print(json.dumps(payload, ensure_ascii=False, indent=2))
180 return 0
181
182
183if __name__ == "__main__":
184 raise SystemExit(main())