"""`wikitool dist export` - build a distributable, contentless copy of this repo's machinery. `export` copies the pipeline's schema/compiler/control-plane layers (types/, tools/, instructions/, the stage contracts) into an empty target, with no kb/ pages, no raw/ content, and no git history - see instructions/setup-instance.md for what happens after. It never calls git. The two binding-but-instance-owned documents under kb/ - each collection's COLLECTION.md and kb/CONVENTIONS.md - cross as `.template` and are adopted by a rename, the same split USER.md/SOUL.md use at the repo root. Three independent exclusion mechanisms feed the plan, for three different shapes of "does not belong in someone else's instance": - Every copied text file passes through `strip_markers()`, which removes any region between `` and ``, markers included - for dev-only *content inside* a file that is otherwise shipped (e.g. a routing line in AGENTS.md). - `instructions/dev/` is pruned from the copy wholesale - for dev-only *whole files* (procedures and the skill that switches an agent into tool-development mode). One-way: nothing reconstructs it in a distributed instance, on purpose - see instructions/dev/ itself for the current contents and AGENTS.md's routing line for what a dev instance sees instead. - Build output under `tools/` is dropped, by directory (`TOOLS_EXCLUDE_DIRS`) where it has one, and by filename (`_is_coverage_output`) where it does not. Not dev-only but *derived*: recomputable, and measured against this repo's own test run rather than the receiving instance's. """ from __future__ import annotations import hashlib import json import os import re import shutil import stat import subprocess import tarfile import tempfile from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Callable, NamedTuple, Optional, Union import typer from chemenu import config, conventions, kb_collections, kb_state, ownership, version as version_mod from chemenu.commands._util import console, fail, rel_path, success, today_iso app = typer.Typer(help="Build a distributable copy of the wiki machinery.") MARKER_START = "" MARKER_END = "" DIST_TEMPLATES_DIR = Path(__file__).resolve().parent.parent / "dist_templates" # Root files copied verbatim (after marker-stripping). INSTALL.md is optional # here: it does not exist until the distribution docs land, and `export` # must not fail just because a later stage of the same repo hasn't shipped # yet. # # The personalization *templates* ship; the filled `USER.md`/`SOUL.md` never # do. This allowlist is what makes that split automatic - a file is copied # because it is named here, so an instance's own personalization is excluded # by construction rather than by a rule someone has to remember. # # `ENVIRONMENT.md.template` rides the same split for the same reason: a # distribution can describe what the file is for, but never what a particular # checkout's harness, MCP servers and remotes are. The filled `ENVIRONMENT.md` # is additionally gitignored, so it is excluded twice over. # # `CLAUDE.md` is harness glue, not a second control plane: Claude Code loads it # and does not load `AGENTS.md`, so it ships for the same reason # `.claude/settings.json` does - a distributed instance running that harness # would otherwise start every session without the control plane. # # `INSTALL-MCP.md` ships beside `INSTALL.md` and for the same reason: the MCP # read server is part of what an instance *has*, even though its dependency is # optional. A distribution whose server is present but undocumented is one # whose operator finds the module by reading the source. # # `DEVELOPMENT.md` is deliberately **absent** from this tuple, unlike every # other root doc above. It documents the release workflow (`version bump` -> # `version release` -> `publish` -> CI tags) and points at `instructions/dev/`, # which this same function excludes wholesale a few lines down - a distributed # instance has no release workflow, no CI and no issue board, so it has # nothing for that document to describe. Do not "fix" this by adding it back: # a root file absent from ROOT_FILES is silently skipped by every export, and # that silence is the correct behaviour here, not a gap. ROOT_FILES = ( "AGENTS.md", "CLAUDE.md", "README.md", "EVALS.md", "INSTALL.md", "INSTALL-MCP.md", ".gitignore", "VERSION", *config.LICENSE_FILES, *config.PERSONALIZATION_TEMPLATES, config.ENVIRONMENT_TEMPLATE, ) # The one part of ROOT_FILES that may not be quietly skipped. Every other entry # copies only `if source.is_file()`, which is right for `INSTALL.md` (it did not # exist until the distribution docs landed) and wrong for a licence: an export # that silently omits it hands the receiving instance the AGPL-covered `tools/` # tree with no licence text, which is a violation the moment that instance is # pushed anywhere public. Missing means the export is broken, not minimal. REQUIRED_ROOT_FILES = config.LICENSE_FILES # Harness-specific session-tracing config: generic machinery (feeds # tools/chemenu/telemetry/ and tools/trace_ingest.py via EVALS.md), not # personal state - unlike `.obsidian/`/`.vscode/`, which are never copied. HOOK_DIRS = (".github/hooks", ".vibe") # tools/ subpaths never copied - build/venv/cache artifacts, not machinery. # `htmlcov/` is coverage.py's HTML report: derived output, and a large tree of # it, measured against the source repo's own test run. `.coveragerc` beside it # *does* ship, the same way `pytest.ini` does - it is configuration, not output. TOOLS_EXCLUDE_DIRS = {".venv", "__pycache__", ".pytest_cache", ".wikitool_session", "htmlcov"} # The rest of coverage's output lands beside the code rather than in a directory # of its own - `.coverage`, `coverage.xml`, and `.coverage..` under a # parallel run - so a directory exclusion cannot reach it. Same argument as # `reports/`: derived, recomputable, and about the source repo rather than about # the instance that would receive it. COVERAGE_OUTPUT_NAMES = frozenset({".coverage", "coverage.xml"}) def _is_coverage_output(filename: str) -> bool: return filename in COVERAGE_OUTPUT_NAMES or filename.startswith(".coverage.") # instructions/dev/ holds stack-development-only procedures and the skill # that switches an agent into tool-development mode - never shipped to a # distributed instance. One-way: there is no `enable-dev`-style command that # reconstructs it afterwards, unlike the marker-block content below. INSTRUCTIONS_EXCLUDE_DIRS = {"dev"} # Fixed by raw/CONTRACT.md's routing table, unlike kb/'s areas (which are # organic - see kb/CONTRACT.md - so `export` does not manufacture them). RAW_SUBDIRS = ("articles", "documents", "notes", "assets") # Stage contracts that are not collections and carry no pages: copied as a # single file each, nothing else from their directory. `kb/` is excluded here # - it is a content stage too, but it has collections underneath it, so its # contract is handled by `build_plan` alongside them rather than as a bare # stage copy. Derived from `ownership.CONTENT_STAGES` rather than listed # again, so the set this loop copies and the set `upstream merge` restores # cannot name a different stage without one of them failing its own test. CONTRACT_ONLY_STAGES = tuple( f"{stage}/CONTRACT.md" for stage in ownership.CONTENT_STAGES if stage != "kb" ) # Single tracked files copied out of an otherwise-untouched, partially-ignored # directory. `.claude/` holds the harness's own session-tracing config # (`settings.json`, tracked) alongside generated skill copies and personal # untracked state (`.claude/skills/`, `.claude/settings.local.json`) - neither # of which belongs in a distribution. Adding `.claude` to HOOK_DIRS would copy # the whole directory, skills included; a single-file entry avoids that # without needing an exclude set HOOK_DIRS doesn't otherwise carry. SINGLE_FILES = (".claude/settings.json",) Content = Union[str, bytes] class PlannedFile(NamedTuple): content: Content executable: bool = False _MARKER_TOKEN_RE = re.compile(re.escape(MARKER_START) + "|" + re.escape(MARKER_END)) # The leading/trailing `\n?` consume the blank line on each side of the # block - the convention is that a marker block always sits as its own # paragraph. Without eating both, a strip leaves two blank lines where the # clean file only ever had one. _MARKER_BLOCK_RE = re.compile( r"\n?" + re.escape(MARKER_START) + r".*?" + re.escape(MARKER_END) + r"\n?", re.DOTALL ) def _validate_markers(text: str, label: str) -> None: """A marker file must be a sequence of well-formed, non-nested start/end pairs. Malformed markers would make `strip_markers` remove either too little or too much, silently - this fails loudly instead.""" depth = 0 for match in _MARKER_TOKEN_RE.finditer(text): if match.group() == MARKER_START: if depth != 0: fail(f"{label}: nested dist:strip-start markers are not supported") depth = 1 else: if depth != 1: fail(f"{label}: dist:strip-end without a matching dist:strip-start") depth = 0 if depth != 0: fail(f"{label}: dist:strip-start without a matching dist:strip-end") def strip_markers(text: str) -> str: """Remove every marked region, markers included. Generic by design: it does not matter what is inside, or how many regions a file has.""" return _MARKER_BLOCK_RE.sub("", text) def _is_executable(path: Path) -> bool: return bool(path.stat().st_mode & stat.S_IXUSR) def _read_planned_file(path: Path, label: str) -> PlannedFile: executable = _is_executable(path) try: text = path.read_text(encoding="utf-8") except UnicodeDecodeError: return PlannedFile(path.read_bytes(), executable) # Marker syntax is an HTML/Markdown comment convention, scoped to .md # files on purpose: applying it to every text file would let the marker # strings themselves - inline here as Python string literals - match as # a region in this file's own source when tools/ gets copied, and eat # the code between them. if path.suffix != ".md": return PlannedFile(text, executable) _validate_markers(text, label) return PlannedFile(strip_markers(text), executable) def _copy_tree( source_root: Path, dest_prefix: str, exclude_dirs: frozenset[str], exclude_file: Optional[Callable[[str], bool]] = None, ) -> dict[str, PlannedFile]: """Every file under source_root, marker-stripped, keyed by its destination-relative path. Excluded directories are pruned during the walk rather than filtered after, so a large `.venv/` is never read. `exclude_file` drops individual files by name, for output that lands beside the code instead of in a directory a prune could catch.""" files: dict[str, PlannedFile] = {} if not source_root.is_dir(): return files for dirpath, dirnames, filenames in os.walk(source_root): dirnames[:] = sorted(d for d in dirnames if d not in exclude_dirs) for filename in sorted(filenames): if exclude_file is not None and exclude_file(filename): continue path = Path(dirpath) / filename relative = path.relative_to(source_root).as_posix() dest_rel = f"{dest_prefix}/{relative}" files[dest_rel] = _read_planned_file(path, dest_rel) return files def _digest(content: Content) -> str: data = content if isinstance(content, bytes) else content.encode("utf-8") return "sha256:" + hashlib.sha256(data).hexdigest() def build_stamp(plan: dict[str, PlannedFile], origin: "Origin") -> str: """The release stamp written into every export. Two jobs. The version and origin fields are what `version check` compares against a release feed - without them an instance cannot tell which stack it is running. The per-file digests are for the update *after* detection: they record what the machinery looked like when it was installed, which is the only way a later upgrade can tell a file the instance edited from one it merely received. Nothing reads them today; writing them now is what keeps that upgrade from needing a format change. """ stamp = { "schema": version_mod.STAMP_SCHEMA, "version": str(version_mod.read_version()), "exported_at": today_iso(), "source_repo": origin.source_repo, "source_commit": origin.source_commit, "release_url": origin.release_url, "update_url": origin.update_url or version_mod.DEFAULT_UPDATE_URL, "files": {relative: _digest(planned.content) for relative, planned in sorted(plan.items())}, } return json.dumps(stamp, indent=2, sort_keys=False) + "\n" class Origin(NamedTuple): """Where this export came from. Supplied by the caller (the release workflow knows the commit and the release URL); `dist export` itself never calls git, so it cannot discover any of it.""" source_repo: Optional[str] = None source_commit: Optional[str] = None release_url: Optional[str] = None update_url: Optional[str] = None def instance_owned_type_stems() -> set[str]: """Type-spec stems whose instances are knowledge pages, and which therefore belong to the instance rather than to the stack. The line is `root:`, and it was already in the frontmatter before anyone drew it: `root: kb` means the type describes a page the instance writes, so its prose, its template and its language are the instance's business. Anything else - `instruction` (`root: repo`), `lint-report` (no `base_dir` at all), `type-spec` itself - describes a stack artifact and ships verbatim. Read from `types/` rather than listed, so an instance adding its own page type gets the same treatment without a code change. """ from chemenu.type_resolver import resolver stems: set[str] = set() for type_path, frontmatter in resolver.list_type_specs(): stem = Path(type_path).stem if stem == "type-spec": continue if not frontmatter.get("base_dir"): continue if (frontmatter.get("root") or "kb") != "kb": continue stems.add(stem) return stems def _plan_types() -> dict[str, PlannedFile]: """`types/`, with the page type-specs re-keyed as templates. Same split as the collection contracts, for the same reason and by the same mechanism: the shipped content is a working default rather than something wrong for the receiver, so the file itself crosses - under a name that has to be adopted before it counts. A type-spec's `.schema.yaml` travels with it, because the two are one type (see types/type-spec.md § Anatomy) and adopting half of it would leave a spec validated by a file it does not own. """ plan = _copy_tree(config.TYPES_DIR, "types", frozenset()) stems = instance_owned_type_stems() if not stems: return plan rekeyed: dict[str, PlannedFile] = {} for relative, planned in plan.items(): name = relative.rsplit("/", 1)[-1] stem = name.split(".", 1)[0] if stem in stems: rekeyed[f"{relative}.template"] = planned else: rekeyed[relative] = planned return rekeyed def build_plan(origin: Optional[Origin] = None) -> dict[str, PlannedFile]: """Every (destination-relative path -> planned file) the export writes.""" plan: dict[str, PlannedFile] = {} missing_licences = [ name for name in REQUIRED_ROOT_FILES if not (config.ROOT / name).is_file() ] if missing_licences: fail( "export would ship code without its licence: " + ", ".join(missing_licences) + " missing from the source tree. Restore them before exporting - a " "distribution carrying tools/ without LICENSE is a copyleft violation " "the moment the receiving instance is published." ) for name in ROOT_FILES: source = config.ROOT / name if source.is_file(): plan[name] = _read_planned_file(source, name) plan.update(_copy_tree(config.INSTRUCTIONS_DIR, "instructions", frozenset(INSTRUCTIONS_EXCLUDE_DIRS))) plan.update(_plan_types()) plan.update(_copy_tree( config.ROOT / "tools", "tools", frozenset(TOOLS_EXCLUDE_DIRS), _is_coverage_output )) for hook_dir in HOOK_DIRS: plan.update(_copy_tree(config.ROOT / hook_dir, hook_dir, frozenset())) # docs/ is stack background - why the stack is built the way it is - and # ships verbatim like instructions/ and types/: it carries no page, no # frontmatter, and (AGENTS.md § File naming) no normative sentence, so # there is nothing instance-owned in it to split off as a .template. plan.update(_copy_tree(config.ROOT / "docs", "docs", frozenset())) # `kb/CONTRACT.md` is stack-owned and ships verbatim; everything beside it # under `kb/` is the instance's own and ships only as a `.template`. That is # the personalization split (`USER.md`/`SOUL.md`) one directory down, and # the reason for it is the same: a distribution can say what the file # decides, never what this instance decided. kb_contract = config.KB_DIR / "CONTRACT.md" if kb_contract.is_file(): plan["kb/CONTRACT.md"] = _read_planned_file(kb_contract, "kb/CONTRACT.md") conventions_template = config.KB_DIR / conventions.CONVENTIONS_TEMPLATE if conventions_template.is_file(): rel = f"kb/{conventions.CONVENTIONS_TEMPLATE}" plan[rel] = _read_planned_file(conventions_template, rel) # A collection contract is instance-owned too, but unlike `USER.md` the # shipped content is not *wrong* for the receiver - it is the profile this # repo's own collections adopted, and a fine starting point. So the file # itself ships, under the template name: one source of truth here, and a # receiving instance that has to rename it before it counts. Keeping a # separate `.template` beside each contract would have meant maintaining two # near-identical copies of the same text, which is the drift AGENTS.md # invariant 8 exists to prevent. for collection in kb_collections.iter_kb_collections(): source = collection / kb_collections.CONTRACT_NAME rel = f"kb/{collection.name}/{kb_collections.CONTRACT_NAME}.template" plan[rel] = _read_planned_file(source, rel) for relative in CONTRACT_ONLY_STAGES: source = config.ROOT / relative if source.is_file(): plan[relative] = _read_planned_file(source, relative) for relative in SINGLE_FILES: source = config.ROOT / relative if source.is_file(): plan[relative] = _read_planned_file(source, relative) for sub in RAW_SUBDIRS: plan[f"raw/{sub}/.gitkeep"] = PlannedFile("") plan["kb/log.md"] = PlannedFile((DIST_TEMPLATES_DIR / "log.md").read_text(encoding="utf-8")) plan["CHANGES.md"] = PlannedFile((DIST_TEMPLATES_DIR / "CHANGES.md").read_text(encoding="utf-8")) # A fresh instance's content is empty, so it is trivially in the shape this # machinery expects - which is exactly what makes the initial declaration # safe to write here rather than leaving it to `migrate baseline`. Only an # instance predating this file has to answer that question by hand. # # `.base`, not the raw `VERSION`: a content shape has no beta channel # (`kb_state.read_kb_version` refuses one), so exporting mid-candidate # still declares the release the content is shaped for, not the candidate # in progress. The stamp below carries the honest, suffix-inclusive value - # the two files answer different questions. plan[kb_state.KB_STATE_FILENAME] = PlannedFile( kb_state.render_kb_state(version_mod.read_version().base, []) ) # Last, so it can digest everything above it. It is the one file in the # export that describes the export rather than being copied into it. plan[version_mod.RELEASE_STAMP_FILENAME] = PlannedFile( build_stamp(plan, origin or Origin()) ) return plan # Content that must never appear in a plan, expressed structurally rather than # by matching text. Three allowlists feed `build_plan`, and each one holds only # because someone remembered the rule when they edited it - nothing re-checks # the result. This does. # # The checks are deliberately structural: a filled personalization file, a kb # page, a raw source, a dev-only instruction. A text-pattern scan (hostnames, # IP literals) was considered and rejected - the project's own host legitimately # appears in INSTALL.md and version.py, so such a scan would either whitelist # the very string it is looking for or cry wolf on every export. # # `COLLECTION.md` and `CONVENTIONS.md` are deliberately *not* allowed through # any more. Both bind, and both are the instance's to write, so they cross the # boundary as `.template` and are adopted by a rename - a plan carrying the # filled name would hand a new instance this one's authoring conventions as # though they were the stack's. # # What counts as machinery under kb/ or raw/ is no longer a second list here: # it is `ownership.is_stack_owned`, the same predicate `upstream merge` and # `upstream verify` restore/check against. Only the export-only stubs # (`ownership.EXPORT_STUB_NAMES`) are allowed here without also being # stack-owned - a merge keeps the *local* copy of those, while export writes a # fresh one regardless of either side, so the two callers genuinely disagree # about them and each keeps its own allowance for that one case. _CONTENT_PREFIXES = ("kb/", "raw/") _INSTANCE_OWNED_KB_FILES = (kb_collections.CONTRACT_NAME, conventions.CONVENTIONS_FILENAME) def find_leaks(plan: dict[str, PlannedFile]) -> list[str]: """Planned paths that carry one instance's own data instead of machinery.""" owned_types = instance_owned_type_stems() leaks: list[str] = [] for relative in sorted(plan): name = relative.rsplit("/", 1)[-1] if name in config.PERSONALIZATION_FILES or name == config.ENVIRONMENT_FILE: leaks.append(f"{relative} (one instance's own personalization)") elif relative.startswith("kb/") and name in _INSTANCE_OWNED_KB_FILES: leaks.append(f"{relative} (this instance's authoring conventions; ship the .template)") elif ( relative.startswith("types/") and not relative.endswith(".template") and name.split(".", 1)[0] in owned_types ): leaks.append(f"{relative} (this instance's page type-spec; ship the .template)") elif relative.startswith("instructions/dev/"): leaks.append(f"{relative} (stack-development only)") elif ( relative.startswith(_CONTENT_PREFIXES) and not ownership.is_stack_owned(relative) and not ownership.is_export_stub(name) ): leaks.append(f"{relative} (wiki content, not machinery)") return leaks def _write_plan(target: Path, plan: dict[str, PlannedFile]) -> None: for relative, planned in plan.items(): dest = target / relative dest.parent.mkdir(parents=True, exist_ok=True) if isinstance(planned.content, bytes): dest.write_bytes(planned.content) else: dest.write_text(planned.content, encoding="utf-8") if planned.executable: dest.chmod(dest.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) @app.command("export") def export_command( target: Path = typer.Argument( ..., help="Directory to write the distribution into. Must not exist, or must be empty." ), dry_run: bool = typer.Option( False, "--dry-run", help="List what would be written, without writing anything." ), source_repo: Optional[str] = typer.Option( None, "--source-repo", help="Repository this export was built from (recorded in the stamp)" ), source_commit: Optional[str] = typer.Option( None, "--source-commit", help="Commit this export was built from (recorded in the stamp)" ), release_url: Optional[str] = typer.Option( None, "--release-url", help="Release page this export ships as (recorded in the stamp)" ), update_url: Optional[str] = typer.Option( None, "--update-url", help="Release feed `version check` should ask (recorded in the stamp)" ), ): """Export a contentless, distributable copy of this repo's machinery: AGENTS.md/README.md (dev-instance-only marker blocks removed), instructions/ (no instructions/dev/), types/, docs/ verbatim, tools/ (no venv/caches), the .github/hooks/+.vibe session-tracing config plus .claude/settings.json, kb/CONTRACT.md plus a COLLECTION.md.template per collection and kb/CONVENTIONS.md.template (no pages, no areas), empty raw/{articles,documents,notes,assets}/, VERSION, the USER.md/SOUL.md personalization templates (never the filled files), and a .wikitool-release.json stamp. The --source-*/--release-url/--update-url options only fill fields in that stamp: `export` never calls git and cannot discover them. See instructions/setup-instance.md for what comes next.""" run_export( target, dry_run=dry_run, origin=Origin( source_repo=source_repo, source_commit=source_commit, release_url=release_url, update_url=update_url, ), ) def run_export(target: Path, dry_run: bool = False, origin: Optional[Origin] = None) -> None: """The export itself, free of Typer's option objects so it can be called directly - by the command above, and by the tests.""" target = target.resolve() if target.exists(): if not target.is_dir(): fail(f"{target} exists and is not a directory.") if any(target.iterdir()): fail(f"{target} is not empty. `dist export` refuses to write into a non-empty directory.") try: plan = build_plan(origin) except version_mod.VersionError as exc: fail(f"{exc} - a distribution must carry the version it ships.") return leaks = find_leaks(plan) if leaks: fail( "export would carry this instance's own data, not just machinery:\n " + "\n ".join(leaks) + "\nThis is an allowlist bug in dist_cmd.py, not something to work " "around - fix the allowlist rather than deleting files from the target." ) return if dry_run: for relative in sorted(plan): typer.echo(f"write {relative}") success(f"Dry run: would write {len(plan)} file(s) to {target}. Nothing written.") return _write_plan(target, plan) success(f"Exported {len(plan)} file(s) to {rel_path(target)}.") # --- dist upgrade ------------------------------------------------------------ # # Apply a release `dist export` produced, rather than merely detecting one # (`version check`). Gitea #7 has the full design; the short version: the # write set is exactly the *new* stamp's `files` block, minus the paths an # export seeds once and the instance owns from then on # (`ownership.is_export_stub`, `ownership.is_upgrade_preserved`), plus the # stamp itself. Every candidate path is classified against the *old* stamp's # recorded digest - unchanged, locally modified, or locally deleted - and a # modified/deleted file is never silently overwritten. This never calls a # release feed; the caller supplies an already-downloaded tree or archive. @dataclass(frozen=True) class FileClassification: """The four-way split of every path `dist upgrade` would touch, plus the fifth direction (`removed`) that has no write set of its own.""" unchanged: list[str] modified: list[str] deleted: list[str] new: list[str] removed: list[str] @property def blocked(self) -> list[str]: """Locally changed paths - modified or deleted - which are never silently overwritten.""" return sorted(self.modified + self.deleted) def _write_candidates(new_files: dict) -> set[str]: """Every path `dist upgrade` may write, from the new stamp's `files` block: everything except the paths an export re-seeds from a blank template every time (`ownership.is_export_stub`) and the paths an export seeds once and the instance owns afterward (`ownership.is_upgrade_preserved`). The release stamp itself is added separately - it is never a member of its own `files` block, see `build_stamp`.""" return { relative for relative in new_files if not ownership.is_export_stub(Path(relative).name) and not ownership.is_upgrade_preserved(relative) } def _classify_files(old_files: dict, new_files: dict) -> FileClassification: candidates = _write_candidates(new_files) recorded_for_candidates = {r: d for r, d in old_files.items() if r in candidates} statuses = kb_state.compare_against_stamp(recorded_for_candidates) unchanged: list[str] = [] modified: list[str] = [] deleted: list[str] = [] new: list[str] = [] for relative in sorted(candidates): if relative not in old_files: new.append(relative) continue status = statuses[relative] if status == kb_state.UNCHANGED: unchanged.append(relative) elif status == kb_state.MODIFIED: modified.append(relative) else: deleted.append(relative) removed = sorted(set(old_files) - set(new_files)) return FileClassification(unchanged, modified, deleted, new, removed) def _verify_sha256_sidecar(archive: Path) -> None: """WARN, never fail, on a missing sidecar - only a corrupted one that *is* present is a reason to stop, per Gitea #7's design table.""" sidecar = archive.with_name(archive.name + ".sha256") if not sidecar.is_file(): console.print( f"[yellow]WARN[/yellow] No {sidecar.name} beside {archive.name} - the archive's " "integrity is not being checked before it is extracted." ) return expected = sidecar.read_text(encoding="utf-8").strip().split()[0:1] actual = hashlib.sha256(archive.read_bytes()).hexdigest() if not expected or expected[0].lower() != actual.lower(): fail( f"{archive.name} does not match {sidecar.name}: expected " f"{expected[0] if expected else '(unreadable)'}, got {actual}. Re-download the " "release archive rather than trusting one that failed its own checksum." ) def _extract_single_top_level_dir(archive: Path, dest: Path) -> Path: """Extract `archive` into `dest` and return the one top-level directory it contained - the shape `.gitea/workflows/release.yml` packs (see its `Build the distribution tarball` step). Refuses anything else rather than guessing which part is the machinery.""" with tarfile.open(archive) as tf: names = [n for n in tf.getnames() if n not in ("", ".")] top_levels = {n.split("/", 1)[0] for n in names} if len(top_levels) != 1: fail( f"{archive.name} does not have exactly one top-level directory (found " f"{len(top_levels)}: {', '.join(sorted(top_levels)) or '(empty archive)'}) - this " "is not the shape a release tarball has, and `dist upgrade` refuses to guess " "which part is the machinery." ) return dest # unreachable: fail() raises typer.Exit try: tf.extractall(dest, filter="data") # noqa: S202 - trusted local archive, path-checked below except TypeError: # Python < 3.12 has no `filter=` argument. Same guard by hand: # refuse any member whose extracted path would land outside dest. resolved_dest = dest.resolve() for member in tf.getmembers(): if not (resolved_dest / member.name).resolve().is_relative_to(resolved_dest): fail( f"{archive.name} contains a path that escapes the extraction directory: " f"{member.name}" ) return dest # unreachable tf.extractall(dest) # noqa: S202 - every member path-checked above return dest / next(iter(top_levels)) @contextmanager def _resolved_source(source: Path): """Yield the directory holding a distribution export: `source` itself if it already is one, or the single top-level directory of a `.tar.gz` extracted into a scratch directory that is cleaned up afterward.""" if source.is_dir(): yield source return if not source.is_file(): fail(f"{source} does not exist.") return _verify_sha256_sidecar(source) with tempfile.TemporaryDirectory(prefix="wikitool-upgrade-") as tmp: yield _extract_single_top_level_dir(source, Path(tmp)) def _git_working_tree_status() -> Optional[str]: """`git status --porcelain` for `config.ROOT`, or None if it is not a git repository at all - which is a valid, if unprotected, state for a tarball instance, not a reason to refuse.""" result = subprocess.run( ["git", "-C", str(config.ROOT), "status", "--porcelain"], capture_output=True, text=True, ) return result.stdout if result.returncode == 0 else None def _report_plan( classification: FileClassification, migration_chain: list["kb_state.Migration"], boundary_crossing: bool, local_version: "version_mod.Version", new_version: "version_mod.Version", ) -> None: console.print(f"{local_version} -> {new_version}") if boundary_crossing: console.print( f"[bold yellow]Crosses a compatibility boundary[/bold yellow] " f"({local_version.compat_key} -> {new_version.compat_key}) - this is not a drop-in " "swap; check the release notes before proceeding." ) console.print( f"{len(classification.unchanged)} unchanged, {len(classification.new)} new, " f"{len(classification.blocked)} locally changed, {len(classification.removed)} removed " "from the release." ) if classification.modified: console.print(f"[bold]Locally modified ({len(classification.modified)}):[/bold]") for relative in classification.modified: console.print(f" - {relative}") if classification.deleted: console.print(f"[bold]Locally deleted ({len(classification.deleted)}):[/bold]") for relative in classification.deleted: console.print(f" - {relative}") if classification.removed: console.print("[dim]No longer part of the release, not written or removed by default:[/dim]") for relative in classification.removed: console.print(f" [dim]- {relative}[/dim]") if migration_chain: console.print( f"[cyan]{len(migration_chain)} migration(s) will be outstanding after this " "upgrade, in this order:[/cyan]" ) for position, migration in enumerate(migration_chain, start=1): console.print(f" {position}. {migration.target} {migration.name} ({migration.kind})") console.print("Report only - `dist upgrade` never runs a migration. See `wikitool migrate status`.") @app.command("upgrade") def upgrade_command( source: Path = typer.Argument( ..., help="An extracted distribution directory, or a release .tar.gz archive" ), dry_run: bool = typer.Option( False, "--dry-run", help="Classify and report, without writing anything" ), keep_local: bool = typer.Option( False, "--keep-local", help="Proceed even with locally changed files - leave each one untouched rather than aborting", ), prune: bool = typer.Option( False, "--prune", help="Also delete files the new release no longer ships, if they are unchanged since install", ), allow_pre: bool = typer.Option( False, "--pre", help="Allow a pre-release (-beta.N) source tree - release.yml never publishes one", ), ): """Apply a stack update `dist export` produced - the write half of `version check`. Never downloads anything: `source` is an already-fetched export directory or `.tar.gz` archive. Writes exactly the new release stamp's `files` block, minus what an export re-seeds every time (`kb/log.md`, `raw/*/.gitkeep`) or seeds once and the instance owns from then on (`.wikitool-kb.json`, `CHANGES.md`), classifying every candidate against the *old* stamp's recorded digest: unchanged files are overwritten silently, new files are created, and a locally modified or deleted file is never silently overwritten - `dist upgrade` aborts unless `--keep-local` says to leave it alone. Reports the migration chain the new machinery would owe without running any of it (there is no `migrate run`). Refuses on a missing local release stamp, a downgrade, a pre-release source without `--pre`, or a dirty working tree. Never touches git. See Gitea #7 and `INSTALL.md` § "Eine Instanz aktualisieren".""" run_upgrade( source, dry_run=dry_run, keep_local=keep_local, prune=prune, allow_pre=allow_pre ) def run_upgrade( source: Path, dry_run: bool = False, keep_local: bool = False, prune: bool = False, allow_pre: bool = False, ) -> None: """The upgrade itself, free of Typer's option objects - see `run_export` for why this split exists.""" try: local_version = version_mod.read_version() except version_mod.VersionError as exc: fail(f"{exc} - this tree has no stack version to upgrade from.") return old_stamp = version_mod.read_stamp() if not old_stamp or not isinstance(old_stamp.get("files"), dict): fail( f"No local {version_mod.RELEASE_STAMP_FILENAME} (or it carries no `files` block). " "Without it, `dist upgrade` cannot tell a file this instance edited from one it " "merely received, and it refuses to guess. A checkout with shared git history takes " "stack updates via `wikitool upstream merge` instead - it has the same information " "as a merge base. A tarball instance that has lost its stamp has no repair path " "today; see Gitea #7 \"Bewusst offen gelassen\"." ) return old_files = old_stamp["files"] try: kb_version = kb_state.read_kb_version() except version_mod.VersionError as exc: fail(str(exc)) return if kb_version is None: fail( f"{kb_state.KB_STATE_FILENAME} is missing - this instance has never declared what " "shape its content is in. Run `wikitool migrate baseline ` before upgrading." ) return outstanding = kb_state.chain(kb_state.load_migrations(), kb_version, local_version.base) if outstanding: fail( f"{len(outstanding)} migration(s) are already outstanding against the installed " f"machinery ({kb_version} -> {local_version}) - `wikitool migrate status` names them. " "Finish them before upgrading further: a machinery swap on top of an unfinished " "migration leaves the corpus in a shape no version describes." ) return tree_status = _git_working_tree_status() if tree_status is None: console.print( "[yellow]WARN[/yellow] Not a git repository (or git is unavailable) - proceeding " "without the dirty-tree check a repository would get." ) elif tree_status.strip(): fail( "Working tree is not clean (`git status --porcelain` printed something). " "`dist upgrade` refuses to start on a dirty tree so a refusal never has to guess " "which changes were already there. Commit or stash first." ) return with _resolved_source(source) as new_root: version_path = new_root / version_mod.VERSION_FILENAME if not version_path.is_file(): fail(f"{rel_path(new_root)} has no VERSION - not a distribution export.") return try: new_version = version_mod.Version.parse(version_path.read_text(encoding="utf-8")) except version_mod.VersionError as exc: fail(str(exc)) return if new_version.is_prerelease and not allow_pre: fail( f"{new_version} is a running candidate (-beta.N). `.gitea/workflows/release.yml` " "never publishes one, so a candidate tree can only come from a dev checkout by " "hand - pass --pre if that is deliberate." ) return if new_version < local_version: fail(f"{new_version} is older than the installed {local_version} - refusing a downgrade.") return if new_version == local_version: success(f"Already at {local_version}. Nothing to do.") return stamp_path = new_root / version_mod.RELEASE_STAMP_FILENAME if not stamp_path.is_file(): fail(f"{rel_path(new_root)} has no {version_mod.RELEASE_STAMP_FILENAME} - not a distribution export.") return try: new_stamp = json.loads(stamp_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError) as exc: fail(f"{version_mod.RELEASE_STAMP_FILENAME} in the source is not readable JSON: {exc}") return new_files = new_stamp.get("files") if isinstance(new_stamp, dict) else None if not isinstance(new_files, dict): fail(f"{version_mod.RELEASE_STAMP_FILENAME} in the source carries no `files` block.") return classification = _classify_files(old_files, new_files) migration_chain = kb_state.chain( kb_state.load_migrations(new_root / "instructions" / kb_state.MIGRATIONS_SUBDIR), kb_version, new_version.base, ) boundary_crossing = local_version.compat_key != new_version.compat_key _report_plan(classification, migration_chain, boundary_crossing, local_version, new_version) # Dry-run's whole purpose is to preview this classification - including # the blocked list - without raising, so it must be checked before the # abort below rather than after: a blocked file must never turn # `--dry-run` into a non-zero exit, or the flag stops being safe to run # freely. if dry_run: success(f"Dry run: would upgrade {local_version} -> {new_version}. Nothing written.") return if classification.blocked and not keep_local: fail( f"{len(classification.blocked)} locally changed file(s) (listed above) would be " "silently overwritten. Pass --keep-local to upgrade anyway and leave every one of " "them untouched, or reconcile them by hand first. Nothing was written." ) return to_write = sorted(classification.unchanged + classification.new) for relative in to_write: src = new_root / relative dst = config.ROOT / relative dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dst) shutil.copy2(stamp_path, config.ROOT / version_mod.RELEASE_STAMP_FILENAME) pruned: list[str] = [] if prune: for relative in classification.removed: digest = old_files.get(relative) if digest is None: continue status = kb_state.compare_against_stamp({relative: digest}).get(relative) if status != kb_state.UNCHANGED: continue target = config.ROOT / relative if target.is_file(): target.unlink() pruned.append(relative) skipped = classification.blocked if keep_local else [] summary = ( f"Upgraded {local_version} -> {new_version}: {len(to_write)} file(s) written" + (f", {len(skipped)} left untouched (--keep-local)" if skipped else "") + (f", {len(pruned)} pruned" if pruned else "") + "." ) if migration_chain: summary += ( f" {len(migration_chain)} migration(s) now outstanding - run `wikitool migrate status`." ) summary += ( " Nothing was committed. Now run, in order: `wikitool instructions sync`, `doctor`, " "`docs verify`, `instructions verify`, `lint` - then restart the agent session." ) success(summary)