60-packaging/gap_work_package.py
#!/usr/bin/env python3
"""Erzeugt und validiert das deterministische Zwei-Archiv-GAP-Arbeitspaket."""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import shutil
import stat
import sys
import tempfile
import zipfile
import zlib
from io import BytesIO
from pathlib import Path, PurePosixPath
from typing import Any, Iterable
from deterministic_zip import FIXED_TIMESTAMP, build_deterministic_zip
class WorkPackageError(RuntimeError):
pass
def _json_bytes(value: Any) -> bytes:
return (json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8")
def _read_json(data: bytes, label: str) -> dict[str, Any]:
try:
value = json.loads(data.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise WorkPackageError(f"{label} ist kein gültiges UTF-8-JSON: {exc}") from exc
if not isinstance(value, dict):
raise WorkPackageError(f"{label} muss ein JSON-Objekt sein")
return value
def _safe_infos(archive: zipfile.ZipFile, label: str) -> list[zipfile.ZipInfo]:
result: list[zipfile.ZipInfo] = []
seen: set[str] = set()
for info in archive.infolist():
name = info.filename
if "\\" in name:
raise WorkPackageError(f"{label}: ZIP-Pfad mit Backslash nicht erlaubt: {name}")
path = PurePosixPath(name)
if not name or path.is_absolute() or ".." in path.parts:
raise WorkPackageError(f"{label}: unsicherer ZIP-Pfad: {name}")
canonical = path.as_posix()
if canonical in seen:
raise WorkPackageError(f"{label}: doppelter ZIP-Pfad: {canonical}")
seen.add(canonical)
mode = (info.external_attr >> 16) & 0o170000
if mode == stat.S_IFLNK:
raise WorkPackageError(f"{label}: symbolischer Link nicht erlaubt: {name}")
result.append(info)
return result
def _open_zip(data: bytes, label: str) -> tuple[zipfile.ZipFile, BytesIO, list[zipfile.ZipInfo]]:
stream = BytesIO(data)
if not zipfile.is_zipfile(stream):
raise WorkPackageError(f"{label}: kein gültiges ZIP")
stream.seek(0)
archive = zipfile.ZipFile(stream)
try:
infos = _safe_infos(archive, label)
damaged = archive.testzip()
if damaged is not None:
raise WorkPackageError(f"{label}: beschädigter ZIP-Eintrag: {damaged}")
except (RuntimeError, zipfile.BadZipFile, zlib.error) as exc:
archive.close()
raise WorkPackageError(f"{label}: beschädigtes ZIP: {exc}") from exc
except Exception:
archive.close()
raise
return archive, stream, infos
def _root_manifest(
archive: zipfile.ZipFile,
infos: list[zipfile.ZipInfo],
label: str,
allow_flat: bool = False,
) -> tuple[str, dict[str, Any]]:
files = [info for info in infos if not info.is_dir()]
candidates = [
info for info in files
if PurePosixPath(info.filename).name == "manifest.json"
and (len(PurePosixPath(info.filename).parts) == 2 or (allow_flat and len(PurePosixPath(info.filename).parts) == 1))
]
if len(candidates) != 1:
raise WorkPackageError(f"{label}: genau ein Root-manifest.json erwartet, gefunden: {len(candidates)}")
candidate_parts = PurePosixPath(candidates[0].filename).parts
root = candidate_parts[0] if len(candidate_parts) == 2 else ""
if root and any(PurePosixPath(info.filename).parts[0] != root for info in infos):
raise WorkPackageError(f"{label}: alle Einträge müssen unter genau einem Repository-Root liegen")
return root, _read_json(archive.read(candidates[0]), f"{label}/manifest.json")
def _parse_frontmatter(text: str, label: str) -> dict[str, str]:
if not text.startswith("---\n"):
raise WorkPackageError(f"{label}: YAML-Frontmatter fehlt")
try:
block = text.split("---\n", 2)[1]
except IndexError as exc:
raise WorkPackageError(f"{label}: YAML-Frontmatter ist nicht abgeschlossen") from exc
fields: dict[str, str] = {}
for line in block.splitlines():
match = re.match(r"^([a-zA-Z0-9_]+):\s*(.*?)\s*$", line)
if match:
fields[match.group(1)] = match.group(2)
return fields
def _inspect_gap(data: bytes) -> dict[str, Any]:
archive, _, infos = _open_zip(data, "gap.zip")
try:
root, manifest = _root_manifest(archive, infos, "gap.zip")
if manifest.get("artifactType") != "semantic-gap" or manifest.get("packageClass") != "installable":
raise WorkPackageError("gap.zip: installierbares semantic-gap erwartet")
expected_root = manifest.get("artifactName")
if not isinstance(expected_root, str) or not expected_root:
repository = manifest.get("repository")
expected_root = repository.get("id") if isinstance(repository, dict) else None
if not isinstance(expected_root, str) or root != expected_root:
raise WorkPackageError(f"gap.zip: unerwarteter Repository-Root: {root}")
active = manifest.get("activeWorkItem")
state = manifest.get("stateMachine")
if not isinstance(active, dict) or not isinstance(state, dict):
raise WorkPackageError("gap.zip: activeWorkItem und stateMachine müssen Objekte sein")
work_id = active.get("id")
work_path = active.get("path")
if not isinstance(work_id, str) or not work_id or not isinstance(work_path, str) or not work_path:
raise WorkPackageError("gap.zip: aktive Work-Item-ID oder Pfad fehlt")
path = PurePosixPath(work_path)
if path.is_absolute() or ".." in path.parts:
raise WorkPackageError(f"gap.zip: unsicherer Work-Item-Pfad: {work_path}")
if state.get("activeWorkItem") != work_id:
raise WorkPackageError("gap.zip: stateMachine und activeWorkItem widersprechen sich")
if active.get("approved") is not True or active.get("status") not in {"ready", "executing"}:
raise WorkPackageError("gap.zip: aktives Work Item muss genehmigt und ready oder executing sein")
work_member = f"{root}/{path.as_posix()}"
prompt_member = f"{root}/PROMPT.md"
names = {info.filename for info in infos if not info.is_dir()}
if work_member not in names or prompt_member not in names:
raise WorkPackageError("gap.zip: aktives Work Item oder stabiler Root-Prompt fehlt")
work_text = archive.read(work_member).decode("utf-8")
base_prompt = archive.read(prompt_member).decode("utf-8")
frontmatter = _parse_frontmatter(work_text, work_member)
if frontmatter.get("id") != work_id or frontmatter.get("approved") != "true" or frontmatter.get("status") not in {"ready", "executing"}:
raise WorkPackageError("gap.zip: Work-Item-Frontmatter widerspricht dem Manifest")
allowed = active.get("allowedTargetRepositories")
if not isinstance(allowed, list) or not allowed or any(not isinstance(item, str) or not item for item in allowed):
raise WorkPackageError("gap.zip: allowedTargetRepositories muss eine nicht leere Stringliste sein")
if len(set(allowed)) != len(allowed):
raise WorkPackageError("gap.zip: allowedTargetRepositories enthält Duplikate")
declared_allowed = frontmatter.get("allowed_target_repositories")
if declared_allowed:
try:
declared = json.loads(declared_allowed)
if (
not isinstance(declared, list)
or not declared
or any(not isinstance(item, str) or not item for item in declared)
or len(set(declared)) != len(declared)
):
raise WorkPackageError(
"gap.zip: allowed_target_repositories muss eine duplikatfreie JSON-Stringliste sein"
)
if set(declared) != set(allowed):
raise WorkPackageError("gap.zip: Allowed Target Scope in Manifest und Work Item widerspricht sich")
except json.JSONDecodeError as exc:
raise WorkPackageError("gap.zip: allowed_target_repositories ist nicht als JSON-Liste lesbar") from exc
generated_prompt = (
base_prompt.rstrip()
+ "\n\n---\n\n# Aktiviertes Work Item\n\n"
+ f"Pfad: `{work_path}`\n\n"
+ work_text.lstrip()
)
return {
"manifest": manifest,
"root": root,
"workItemId": work_id,
"workItemPath": work_path,
"workItemStatus": active["status"],
"allowedTargets": allowed,
"generatedPrompt": generated_prompt.encode("utf-8"),
}
finally:
archive.close()
def _inspect_repository(data: bytes, expected_name: str, label: str) -> dict[str, Any]:
archive, _, infos = _open_zip(data, label)
try:
root, manifest = _root_manifest(archive, infos, label, allow_flat=True)
identity = {str(manifest.get("artifactType", "")), str(manifest.get("artifactName", ""))}
if expected_name not in identity:
raise WorkPackageError(
f"{label}: Repository-Identität stimmt nicht; erwartet {expected_name}, erhalten "
f"{manifest.get('artifactType')}/{manifest.get('artifactName')}"
)
if manifest.get("artifactType") == "semantic-gap":
raise WorkPackageError(f"{label}: semantic-gap darf nicht als Repository eingeordnet werden")
if root and root != expected_name:
raise WorkPackageError(f"{label}: unerwarteter Repository-Root: {root}; erwartet {expected_name}")
return {"root": root, "rootForm": "wrapped" if root else "flat", "manifest": manifest}
finally:
archive.close()
def _parse_named_archives(values: Iterable[str], role: str) -> list[tuple[str, Path]]:
parsed: list[tuple[str, Path]] = []
seen: set[str] = set()
for value in values:
if "=" not in value:
raise WorkPackageError(f"{role}: NAME=PFAD erwartet: {value}")
name, raw_path = value.split("=", 1)
if not re.fullmatch(r"[A-Za-z0-9._-]+", name) or name in {"gap", "PROMPT", "package-manifest"}:
raise WorkPackageError(f"{role}: ungültiger Repositoryname: {name}")
if name in seen:
raise WorkPackageError(f"{role}: Repository doppelt angegeben: {name}")
seen.add(name)
path = Path(raw_path).expanduser()
if not path.is_file():
raise WorkPackageError(f"{role}: Archiv nicht gefunden: {path}")
parsed.append((name, path))
return parsed
def _load_filter_config(path: Path | None) -> dict[str, Any]:
if path is None:
return {"schemaVersion": "1.0", "filters": {}}
if not path.is_file():
raise WorkPackageError(f"Uploadfilter-Konfiguration nicht gefunden: {path}")
config = _read_json(path.read_bytes(), str(path))
if config.get("schemaVersion") != "1.0" or not isinstance(config.get("filters"), dict):
raise WorkPackageError("Uploadfilter-Konfiguration benötigt schemaVersion 1.0 und filters")
return config
def _filter_prefix(pattern: str, label: str) -> str:
if not isinstance(pattern, str):
raise WorkPackageError(f"{label}: Ausschlussmuster muss ein String sein: {pattern!r}")
raw = pattern[1:] if pattern.startswith("!") else pattern
prefix = raw[:-3].rstrip("/") if raw.endswith("/**") else raw
path = PurePosixPath(prefix)
if not prefix or "*" in prefix or path.is_absolute() or ".." in path.parts or "\\" in prefix:
raise WorkPackageError(f"{label}: unsicheres Ausschlussmuster: {pattern}")
return path.as_posix()
def _matches_pattern(path: str, pattern: str, label: str) -> bool:
prefix = _filter_prefix(pattern, label)
return path == prefix or path.startswith(prefix + "/")
def _is_excluded(path: str, patterns: list[str], label: str) -> bool:
"""Apply ordered exclusions and `!`-exceptions deterministically."""
excluded = False
for pattern in patterns:
if _matches_pattern(path, pattern, label):
excluded = not pattern.startswith("!")
return excluded
def _filter_repository_copy(
repository: str,
data: bytes,
config: dict[str, Any],
) -> tuple[bytes, dict[str, Any] | None]:
raw_rule = config["filters"].get(repository)
if raw_rule is None:
return data, None
if not isinstance(raw_rule, dict):
raise WorkPackageError(f"Uploadfilter {repository}: Regel muss ein Objekt sein")
required_strings = ("ruleVersion", "reason", "applyTo")
if any(not isinstance(raw_rule.get(key), str) or not raw_rule[key] for key in required_strings):
raise WorkPackageError(f"Uploadfilter {repository}: Regelversion, Begründung oder Geltungsbereich fehlt")
if raw_rule["applyTo"] != "temporary-upload-copy-only":
raise WorkPackageError(f"Uploadfilter {repository}: unzulässiger Geltungsbereich")
preserve = raw_rule.get("preserve")
if not isinstance(preserve, list) or any(not isinstance(item, str) or not item for item in preserve):
raise WorkPackageError(f"Uploadfilter {repository}: preserve muss eine Stringliste sein")
repository_info = _inspect_repository(data, repository, f"Source {repository}")
pattern_key = "exclude" if repository_info["rootForm"] == "wrapped" else "flatRootEquivalent"
patterns = raw_rule.get(pattern_key)
if not isinstance(patterns, list) or not patterns or any(not isinstance(item, str) for item in patterns):
raise WorkPackageError(f"Uploadfilter {repository}: {pattern_key} muss eine nicht leere Stringliste sein")
for pattern in patterns:
_filter_prefix(pattern, f"Uploadfilter {repository}")
archive, _, infos = _open_zip(data, f"Source {repository}")
try:
entries: dict[str, bytes] = {}
modes: dict[str, int] = {}
excluded: list[str] = []
for info in infos:
if info.is_dir():
continue
name = PurePosixPath(info.filename).as_posix()
if _is_excluded(name, patterns, f"Uploadfilter {repository}"):
excluded.append(name)
continue
entries[name] = archive.read(info)
modes[name] = (info.external_attr >> 16) & 0o777 or 0o644
finally:
archive.close()
stream = BytesIO()
with zipfile.ZipFile(stream, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as filtered:
for name in sorted(entries, key=lambda value: value.encode("utf-8")):
info = zipfile.ZipInfo(name, FIXED_TIMESTAMP)
info.external_attr = (modes[name] & 0xFFFF) << 16
info.compress_type = zipfile.ZIP_DEFLATED
filtered.writestr(info, entries[name], compress_type=zipfile.ZIP_DEFLATED, compresslevel=9)
filtered_bytes = stream.getvalue()
check, _, check_infos = _open_zip(filtered_bytes, f"Gefilterte Source {repository}")
try:
remaining = [
info.filename for info in check_infos
if _is_excluded(info.filename, patterns, f"Uploadfilter {repository}")
]
finally:
check.close()
if remaining:
raise WorkPackageError(f"Uploadfilter {repository}: Negativprüfung fehlgeschlagen: {remaining}")
_inspect_repository(filtered_bytes, repository, f"Gefilterte Source {repository}")
evidence = {
"applyTo": raw_rule["applyTo"],
"excludedPaths": sorted(excluded, key=lambda value: value.encode("utf-8")),
"excludedPatterns": patterns,
"preserve": preserve,
"reason": raw_rule["reason"],
"repository": repository,
"rootForm": repository_info["rootForm"],
"ruleVersion": raw_rule["ruleVersion"],
}
return filtered_bytes, evidence
def _validate_filter_evidence(sources: dict[str, bytes], evidence: Any) -> list[dict[str, Any]]:
if evidence is None:
return []
if not isinstance(evidence, list):
raise WorkPackageError("package-manifest.json: uploadFilters muss eine Liste sein")
validated: list[dict[str, Any]] = []
repositories: set[str] = set()
for item in evidence:
if not isinstance(item, dict):
raise WorkPackageError("package-manifest.json: Uploadfilter-Evidence muss ein Objekt sein")
repository = item.get("repository")
patterns = item.get("excludedPatterns")
excluded_paths = item.get("excludedPaths")
if not isinstance(repository, str) or repository not in sources or repository in repositories:
raise WorkPackageError("package-manifest.json: ungültiges oder doppeltes Uploadfilter-Repository")
repositories.add(repository)
if item.get("applyTo") != "temporary-upload-copy-only" or item.get("rootForm") not in {"wrapped", "flat"}:
raise WorkPackageError(f"Uploadfilter {repository}: Geltungsbereich oder Rootform ungültig")
if not isinstance(item.get("ruleVersion"), str) or not item["ruleVersion"] or not isinstance(item.get("reason"), str) or not item["reason"]:
raise WorkPackageError(f"Uploadfilter {repository}: Regelversion oder Begründung fehlt")
if not isinstance(item.get("preserve"), list) or any(not isinstance(value, str) or not value for value in item["preserve"]):
raise WorkPackageError(f"Uploadfilter {repository}: preserve ungültig")
if not isinstance(patterns, list) or not patterns or any(not isinstance(value, str) for value in patterns):
raise WorkPackageError(f"Uploadfilter {repository}: Ausschlussmuster fehlen")
if not isinstance(excluded_paths, list) or any(not isinstance(value, str) for value in excluded_paths):
raise WorkPackageError(f"Uploadfilter {repository}: ausgeschlossene Pfade ungültig")
if excluded_paths != sorted(set(excluded_paths), key=lambda value: value.encode("utf-8")):
raise WorkPackageError(f"Uploadfilter {repository}: ausgeschlossene Pfade sind nicht eindeutig sortiert")
archive, _, infos = _open_zip(sources[repository], f"Gefilterte Source {repository}")
try:
actual_paths = {info.filename for info in infos}
finally:
archive.close()
actual_root_form = _inspect_repository(
sources[repository], repository, f"Gefilterte Source {repository}"
)["rootForm"]
if item["rootForm"] != actual_root_form:
raise WorkPackageError(f"Uploadfilter {repository}: deklarierte Rootform stimmt nicht")
for pattern in patterns:
_filter_prefix(pattern, f"Uploadfilter {repository}")
if any(_is_excluded(path, patterns, f"Uploadfilter {repository}") for path in actual_paths):
raise WorkPackageError(f"Uploadfilter {repository}: Negativprüfung fehlgeschlagen")
if any(
not _is_excluded(path, patterns, f"Uploadfilter {repository}")
for path in excluded_paths
):
raise WorkPackageError(f"Uploadfilter {repository}: ausgeschlossener Pfad passt zu keinem Muster")
if any(path in actual_paths for path in excluded_paths):
raise WorkPackageError(f"Uploadfilter {repository}: ausgeschlossener Pfad ist im Paket enthalten")
validated.append(item)
return validated
def _identity(manifest: dict[str, Any]) -> dict[str, str]:
result: dict[str, str] = {}
for key in ("artifactType", "artifactName", "packageClass", "version"):
value = manifest.get(key)
if isinstance(value, str) and value:
result[key] = value
return result
def _repository_roles(repositories: dict[str, bytes], role: str) -> list[dict[str, Any]]:
result: list[dict[str, Any]] = []
for name in sorted(repositories, key=lambda value: value.encode("utf-8")):
inspected = _inspect_repository(repositories[name], name, f"{role} {name}")
result.append({
"archive": f"{name}.zip",
"identity": _identity(inspected["manifest"]),
"name": name,
"role": "immutable-source" if role == "Source" else "mutable-target",
"root": inspected["root"] or ".",
"rootForm": inspected["rootForm"],
})
return result
def _package_manifest(
gap: dict[str, Any],
sources: dict[str, bytes],
targets: dict[str, bytes],
upload_filters: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
source_files = ["PROMPT.md"] + [f"{name}.zip" for name in sorted(sources, key=lambda value: value.encode("utf-8"))]
target_files = ["gap.zip"] + [f"{name}.zip" for name in sorted(targets, key=lambda value: value.encode("utf-8"))]
manifest = {
"activeWorkItem": {
"id": gap["workItemId"],
"path": gap["workItemPath"],
"status": gap["workItemStatus"],
},
"allowedTargetRepositories": gap["allowedTargets"],
"artifactType": "gap-two-archive-work-package",
"integrityManifest": "SHA256SUMS.txt",
"packageVersion": "2",
"repositoryRoles": {
"semanticGap": {
"archive": "gap.zip",
"identity": _identity(gap["manifest"]),
"role": "mutable-process-control",
"root": gap["root"],
"rootForm": "wrapped",
},
"sources": _repository_roles(sources, "Source"),
"targets": _repository_roles(targets, "Target"),
},
"roundtrip": {
"returnChangedRepositoriesIndividually": True,
"returnUpdatedSemanticGapArchive": True,
"returnUpdatedTargetsArchive": False,
},
"rules": {
"gapInsideTargets": True,
"oneRunOneActiveWorkItem": True,
"reportNextRequiredFiles": True,
"reportProgress": True,
"sourceImmutable": True,
"targetChangesRequireActiveWorkItemAllowance": True,
"targetMutationControlledByActiveWorkItem": True,
},
"schemaVersion": "1.0",
"sourcesArchive": "sources.zip",
"sourcesContains": source_files,
"targetsArchive": "targets.zip",
"targetsContains": target_files,
}
if upload_filters:
manifest["uploadFilters"] = upload_filters
return manifest
def _checksum_bytes(
artifacts: dict[str, bytes],
source_entries: dict[str, bytes],
target_entries: dict[str, bytes],
) -> bytes:
values = {
"package-manifest.json": artifacts["package-manifest.json"],
"sources.zip": artifacts["sources.zip"],
"targets.zip": artifacts["targets.zip"],
}
values.update({f"sources.zip/{name}": data for name, data in source_entries.items()})
values.update({f"targets.zip/{name}": data for name, data in target_entries.items()})
lines = [
f"{hashlib.sha256(values[name]).hexdigest()} {name}\n"
for name in sorted(values, key=lambda value: value.encode("utf-8"))
]
return "".join(lines).encode("utf-8")
def _build_from_bytes(
gap_bytes: bytes,
sources: dict[str, bytes],
targets: dict[str, bytes],
filter_config: dict[str, Any] | None = None,
) -> tuple[dict[str, bytes], dict[str, Any]]:
gap = _inspect_gap(gap_bytes)
for name, data in sources.items():
_inspect_repository(data, name, f"Source {name}")
for name, data in targets.items():
_inspect_repository(data, name, f"Target {name}")
if name not in gap["allowedTargets"]:
raise WorkPackageError(f"Target {name} liegt nicht im Allowed Target Scope des aktiven Work Items")
packaged_sources: dict[str, bytes] = {}
filter_evidence: list[dict[str, Any]] = []
config = filter_config or {"schemaVersion": "1.0", "filters": {}}
for name, data in sources.items():
packaged, evidence = _filter_repository_copy(name, data, config)
packaged_sources[name] = packaged
if evidence is not None:
filter_evidence.append(evidence)
packaged_targets: dict[str, bytes] = {}
for name, data in targets.items():
# Targets are transfer copies as well. Keep the local original intact,
# but do not transport generated Runtime state or recursive repository
# copies back as mutable target content.
packaged, _ = _filter_repository_copy(name, data, config)
packaged_targets[name] = packaged
filter_evidence.sort(key=lambda item: item["repository"].encode("utf-8"))
source_entries = {"PROMPT.md": gap["generatedPrompt"]}
source_entries.update({f"{name}.zip": data for name, data in packaged_sources.items()})
target_entries = {"gap.zip": gap_bytes}
target_entries.update({f"{name}.zip": data for name, data in packaged_targets.items()})
manifest = _package_manifest(gap, packaged_sources, packaged_targets, filter_evidence)
artifacts = {
"package-manifest.json": _json_bytes(manifest),
"sources.zip": build_deterministic_zip(source_entries),
"targets.zip": build_deterministic_zip(target_entries),
}
artifacts["SHA256SUMS.txt"] = _checksum_bytes(artifacts, source_entries, target_entries)
return artifacts, manifest
def create_work_package(
gap_path: Path,
source_specs: Iterable[str],
target_specs: Iterable[str],
output_dir: Path,
filter_config_path: Path | None = None,
) -> dict[str, Any]:
if not gap_path.is_file():
raise WorkPackageError(f"gap.zip nicht gefunden: {gap_path}")
source_paths = _parse_named_archives(source_specs, "Source")
target_paths = _parse_named_archives(target_specs, "Target")
if not target_paths:
raise WorkPackageError("Mindestens ein Target-Repository ist erforderlich")
gap_bytes = gap_path.read_bytes()
sources = {name: path.read_bytes() for name, path in source_paths}
targets = {name: path.read_bytes() for name, path in target_paths}
filter_config = _load_filter_config(filter_config_path)
artifacts, manifest = _build_from_bytes(gap_bytes, sources, targets, filter_config)
_validate_work_package_bytes(artifacts)
output_dir = output_dir.expanduser().resolve()
if output_dir.exists():
raise WorkPackageError(f"Ausgabeziel existiert bereits und wird nicht überschrieben: {output_dir}")
output_dir.parent.mkdir(parents=True, exist_ok=True)
staging = Path(tempfile.mkdtemp(prefix=f".{output_dir.name}.staging-", dir=output_dir.parent))
try:
for name, data in artifacts.items():
(staging / name).write_bytes(data)
validate_work_package(staging)
if output_dir.exists():
raise WorkPackageError(f"Ausgabeziel wurde während der Erzeugung angelegt: {output_dir}")
staging.replace(output_dir)
finally:
if staging.exists():
shutil.rmtree(staging)
validate_work_package(output_dir)
return manifest
def _root_file_entries(data: bytes, label: str) -> dict[str, bytes]:
archive, _, infos = _open_zip(data, label)
try:
if any(info.is_dir() or len(PurePosixPath(info.filename).parts) != 1 for info in infos):
raise WorkPackageError(f"{label}: ausschließlich Root-Dateien erlaubt")
return {info.filename: archive.read(info) for info in infos}
finally:
archive.close()
def _validate_work_package_bytes(artifacts: dict[str, bytes]) -> dict[str, Any]:
required = {"package-manifest.json", "SHA256SUMS.txt", "sources.zip", "targets.zip"}
if set(artifacts) != required:
raise WorkPackageError(f"Arbeitspaket: exakt {sorted(required)} erwartet, erhalten {sorted(artifacts)}")
actual_manifest = _read_json(artifacts["package-manifest.json"], "package-manifest.json")
source_entries = _root_file_entries(artifacts["sources.zip"], "sources.zip")
target_entries = _root_file_entries(artifacts["targets.zip"], "targets.zip")
if "PROMPT.md" not in source_entries or any(name != "PROMPT.md" and not name.endswith(".zip") for name in source_entries):
raise WorkPackageError("sources.zip: PROMPT.md und ausschließlich Repository-ZIPs erwartet")
if "gap.zip" not in target_entries or any(name != "gap.zip" and not name.endswith(".zip") for name in target_entries):
raise WorkPackageError("targets.zip: gap.zip und ausschließlich Repository-ZIPs erwartet")
sources = {name[:-4]: data for name, data in source_entries.items() if name.endswith(".zip")}
targets = {name[:-4]: data for name, data in target_entries.items() if name.endswith(".zip") and name != "gap.zip"}
gap = _inspect_gap(target_entries["gap.zip"])
if source_entries["PROMPT.md"] != gap["generatedPrompt"]:
raise WorkPackageError("sources.zip: PROMPT.md entspricht nicht Grundvertrag plus aktivem Work Item")
for name, data in sources.items():
_inspect_repository(data, name, f"Source {name}")
for name, data in targets.items():
_inspect_repository(data, name, f"Target {name}")
if name not in gap["allowedTargets"]:
raise WorkPackageError(f"Target {name} liegt nicht im Allowed Target Scope des aktiven Work Items")
upload_filters = _validate_filter_evidence(sources, actual_manifest.get("uploadFilters"))
expected_manifest = _package_manifest(gap, sources, targets, upload_filters)
if actual_manifest != expected_manifest:
raise WorkPackageError("package-manifest.json stimmt nicht mit dem Paketinhalt überein")
expected_checksums = _checksum_bytes(artifacts, source_entries, target_entries)
if artifacts["SHA256SUMS.txt"] != expected_checksums:
raise WorkPackageError("SHA256SUMS.txt stimmt nicht mit dem Paketinhalt überein")
return expected_manifest
def validate_work_package(path: Path) -> dict[str, Any]:
if not path.is_dir():
raise WorkPackageError(f"Arbeitspaketverzeichnis nicht gefunden: {path}")
required = {"package-manifest.json", "SHA256SUMS.txt", "sources.zip", "targets.zip"}
actual = {item.name for item in path.iterdir()}
if actual != required or any(not (path / name).is_file() for name in required):
raise WorkPackageError(f"Arbeitspaket: exakt {sorted(required)} erwartet, erhalten {sorted(actual)}")
artifacts = {name: (path / name).read_bytes() for name in required}
return _validate_work_package_bytes(artifacts)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
create = subparsers.add_parser("create", help="Arbeitspaket erzeugen und validieren")
create.add_argument("--gap", required=True, type=Path)
create.add_argument("--source", action="append", default=[], metavar="NAME=PFAD")
create.add_argument("--target", action="append", default=[], metavar="NAME=PFAD")
create.add_argument("--filter-config", type=Path)
create.add_argument("--output-dir", required=True, type=Path)
validate = subparsers.add_parser("validate", help="Bestehendes Arbeitspaket validieren")
validate.add_argument("package", type=Path)
args = parser.parse_args()
try:
if args.command == "create":
manifest = create_work_package(args.gap, args.source, args.target, args.output_dir, args.filter_config)
print(f"Arbeitspaket erzeugt: {args.output_dir} ({manifest['activeWorkItem']['id']})")
else:
manifest = validate_work_package(args.package)
print(f"Arbeitspaket gültig: {args.package} ({manifest['activeWorkItem']['id']})")
return 0
except (OSError, WorkPackageError, ValueError, zipfile.BadZipFile) as exc:
print(f"FEHLER: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())