feat(site): store transformed assets as depot tasks

docs/stages.md

@@ -326,3 +326,20 @@ The codes of items that are `generic`, `unsupported` or `failed`, and of limits
 ## ui.pack
 
 `ui.pack` assembles the selected APK UI key from its object table, including referenced font objects. It writes `ui.json` with native roots, prefab/controller lists and named animation bindings. The product layout groups its manifests into the UI library index. See [UI libraries](ui.md).
+
+## site.asset
+
+`site.asset` prepares one content-addressed file for a static site. Its subject is `<input sha256>.<extension>`;
+its input role is `source`. The task key includes the input content, transformation parameters and the installed
+encoder implementation. Input names and storage locations do not affect the result.
+
+- `encoding`: `gzip` (default), `br` or `none`. A compressible file uses the encoded form only when it is smaller.
+- `split`: JSON byte threshold (default 524288); `0` keeps every file whole. Objects with at least two supported
+  top-level keys are split into value files that reconstruct the exact minified text.
+- `omit`: JSON key paths to remove before minification, for example `[["metadata", "description"]]`. The default
+  empty list preserves every key. Paths are explicit; native component fields retain their names and values.
+
+The result's `entry.json` is a product file entry, either `{asset, size, stored?}` or `{parts, size}`. Asset paths
+use the decoded content id; compressed bytes have their own depot content id. The result records both byte forms,
+and layout entries select the stored form. Replaying a task reuses its result; changing a parameter or encoder
+version gives a different task key.

src/nnnotes/cli/assets.py

@@ -140,6 +140,7 @@ def matches(self, stage: str) -> bool:
     StageSource("story.host", "nnnotes.pipeline.storyhost:HostStage", output=True, product=True),
     StageSource("story", "nnnotes.pipeline.story:StoryStage", output=True, product=True),
     StageSource("ui.pack", "nnnotes.pipeline.ui:UiStage", output=True, product=True),
+    StageSource("site.asset", "nnnotes.pipeline.siteassets:AssetStage"),
     StageSource("view.", "nnnotes.pipeline.views:stages", output=True),
     StageSource("view.voices", "nnnotes.pipeline.voices:VoiceStage", output=True),
 )

src/nnnotes/cli/site.py

@@ -5,8 +5,8 @@
 import sys
 from pathlib import Path
 
+from ..formats.compression import DEFAULT_ENCODING, ENCODINGS
 from ..formats.encoders import DEFAULT_AUDIO_FORMAT, WEB_AUDIO
-from ..site.compress import DEFAULT_ENCODING, ENCODINGS
 from ..source.config import ConfigError, use
 from ..source.session import Session
 from .common import (

src/nnnotes/pipeline/siteassets.py

@@ -0,0 +1,166 @@
+"""Site asset tasks: JSON projection and minification, splitting, and stored encodings.
+
+Each subject is an input content id and extension. The result holds the decoded content, its encoded form when
+smaller, and an entry document. A layout places these artifact bytes without decoding or recompressing them.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import zlib
+from importlib import metadata
+
+from ..core import jsonio
+from ..formats import compression
+from . import contract
+from .contract import IncompatibleTask, Input
+from .stages import Output, Stage
+
+ASSET = "site.asset"
+SPLIT_MIN_BYTES = 512 * 1024
+_SUBJECT = re.compile(r"([0-9a-f]{64})\.([a-z0-9]+)")
+_SPLIT_KEY = re.compile(r"[A-Za-z0-9_$.\-]+")
+
+
+def subject(inp: Input, extension: str) -> str:
+    value = f"{inp.sha256}.{extension}"
+    if not _SUBJECT.fullmatch(value):
+        raise ValueError(f"invalid site asset extension: {extension!r}")
+    return value
+
+
+def _extension(value: str) -> str:
+    match = _SUBJECT.fullmatch(value)
+    if not match:
+        raise ValueError(f"invalid site asset subject: {value!r}")
+    return match[2]
+
+
+def _json(value) -> bytes:
+    return jsonio.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+
+
+def atoms(extension: str, encoding: str) -> dict:
+    used = {"json.minify": "nnnotes-json-minify/1"} if extension == "json" else {}
+    if extension in compression.COMPRESSIBLE:
+        if encoding == "gzip":
+            used["asset.gzip"] = f"nnnotes-gzip/1:zlib/{zlib.ZLIB_RUNTIME_VERSION}"
+        elif encoding == "br":
+            used["asset.brotli"] = f"nnnotes-brotli/1:brotli/{metadata.version('brotli')}"
+    return used
+
+
+class AssetStage(Stage):
+    """One source file transformed for the site, shared across logical paths and products."""
+
+    name, version = ASSET, 1
+    PARAMS = {"encoding": "gzip", "split": SPLIT_MIN_BYTES, "omit": []}
+
+    def normalize(self, params):
+        p = super().normalize(params)
+        compression.check(p["encoding"])
+        if type(p["split"]) is not int or p["split"] < 0:
+            raise ValueError("site.asset split must be a nonnegative byte count")
+        paths = p["omit"]
+        if not isinstance(paths, list) or any(
+            not isinstance(parts, list) or not parts or any(not isinstance(k, str) or not k for k in parts)
+            for parts in paths
+        ):
+            raise ValueError("site.asset omit must be an array of nonempty string paths")
+        p["omit"] = [list(parts) for parts in sorted({tuple(parts) for parts in paths})]
+        return p
+
+    def subjects(self, env):
+        return sorted(env.facts.get("site_assets", {}))
+
+    def inputs(self, subject, env):
+        _extension(subject)
+        inp = env.fact("site_assets")[subject]
+        if not subject.startswith(inp.sha256 + "."):
+            raise ValueError("site asset subject differs from its input content id")
+        return [Input("source", inp.sha256, inp.size, inp.name, inp.locators)]
+
+    def uses(self, subject, env):
+        return atoms(_extension(subject), env.params["encoding"])
+
+    def run(self, task, depot):
+        ext = _extension(task.subject)
+        if task.atoms != atoms(ext, task.params["encoding"]):
+            raise IncompatibleTask("site asset encoder implementations differ from the task")
+        data = depot.input_bytes(task.input("source"))
+        parts = None
+        if ext == "json":
+            doc = json.loads(data)
+            for path in task.params["omit"]:
+                parent = doc
+                for key in path[:-1]:
+                    parent = parent.get(key) if isinstance(parent, dict) else None
+                if isinstance(parent, dict):
+                    parent.pop(path[-1], None)
+            data = _json(doc)
+            threshold = task.params["split"]
+            if threshold and len(data) > threshold and isinstance(doc, dict) and len(doc) > 1:
+                if all(_SPLIT_KEY.fullmatch(k) for k in doc):
+                    parts = [(k, _json(v)) for k, v in doc.items()]
+                    joined = b"{" + b",".join(_json(k) + b":" + v for k, v in parts) + b"}"
+                    if joined != data:
+                        raise ValueError("site JSON parts differ from the complete document")
+        elif task.params["omit"]:
+            raise ValueError("site asset omit applies to JSON only")
+
+        artifacts, placed = [], {}
+
+        def emit(role, body, extension, kind, **facts):
+            aid = contract.artifact_id(task.id, role)
+            artifacts.append(
+                contract.artifact(aid, depot.add(body, extension), contract.provenance(task), {"kind": kind, **facts})
+            )
+            return aid
+
+        def store(body):
+            sha = contract.sha256(body)
+            if sha in placed:
+                return dict(placed[sha])
+            path = f"assets/{sha}.{ext}"
+            encoded = None
+            if task.params["encoding"] != "none" and ext in compression.COMPRESSIBLE:
+                candidate = compression.encode(task.params["encoding"], body)
+                if len(candidate) < len(body):
+                    encoded = candidate
+            entry = {"asset": path, "size": len(body)}
+            if encoded is None:
+                emit(sha, body, ext, "site.asset", path=path)
+            else:
+                suffix = compression.SUFFIXES[task.params["encoding"]]
+                entry.update(asset=path + suffix, stored=len(encoded))
+                emit(sha, body, ext, "site.decoded")
+                emit(sha + suffix, encoded, suffix[1:], "site.asset", path=entry["asset"])
+            placed[sha] = entry
+            return dict(entry)
+
+        if parts is None:
+            entry = store(data)
+        else:
+            rows = []
+            for key, body in parts:
+                piece = store(body)
+                rows.append([key, piece["asset"], piece["size"], *([piece["stored"]] if "stored" in piece else [])])
+            entry = {"size": len(data), "parts": rows}
+        emit("entry.json", contract.encode(entry), "json", "site.entry")
+        return Output(artifacts)
+
+
+def entry(result, depot) -> dict:
+    """The product file entry represented by a committed site asset task."""
+    artifact = next(a for a in result["artifacts"] if a["semantics"]["kind"] == "site.entry")
+    return contract.loads(depot.read(artifact["content"]["sha256"]))
+
+
+def placements(result) -> list[dict]:
+    """Layout entries for the task's stored bytes, with decoded content ids in their paths."""
+    return [
+        {"path": a["semantics"]["path"], "sha256": a["content"]["sha256"], "size": a["content"]["size"], "id": a["id"]}
+        for a in result["artifacts"]
+        if a["semantics"]["kind"] == "site.asset"
+    ]

src/nnnotes/site/sitefiles.py

@@ -25,11 +25,11 @@
 from pathlib import Path
 
 from ..core import cache, jsonio
+from ..formats import compression as compress
+from ..formats.compression import DEFAULT_ENCODING
 from ..formats.shader import PLAYER_PLATFORM as SHADER_PLATFORM
 from ..formats.shader import PLAYER_TYPE as SHADER_TYPE
 from ..source.config import Config, ConfigError
-from . import compress
-from .compress import DEFAULT_ENCODING
 
 SITE_FORMAT = 2  # charts.json "format"
 MANIFEST_FORMAT = 3  # chart manifest "format" (3: entries may be stored encoded)

src/nnnotes/site/siteindex.py

@@ -7,8 +7,8 @@
 from pathlib import Path
 
 from ..content.score import DIFFICULTIES
+from ..formats import compression as compress
 from ..source import languages
-from . import compress
 from .sitefiles import (
     CHARTS_INDEX,
     MODELS_DIR,

src/nnnotes/site/web.py

@@ -82,12 +82,12 @@
 from ..content import live, livenotes, liveoptions, livescene
 from ..content.score import DIFFICULTIES, master_table
 from ..core import cache, jsonio
+from ..formats import compression as compress
 from ..formats import cri
+from ..formats.compression import DEFAULT_ENCODING
 from ..formats.encoders import DEFAULT_AUDIO_FORMAT, WEB_AUDIO, mp4_priming, web_encode
 from ..source import languages
 from ..source.config import Config, ConfigError, tool, usable_cpus, use
-from . import compress
-from .compress import DEFAULT_ENCODING
 from .sitefiles import (
     MANIFEST_FORMAT,
     READ_SET_SCRIPT,

tests/pipeline/test_siteassets.py

@@ -0,0 +1,137 @@
+"""Site asset bytes, task identities and placement records over synthetic content."""
+
+import json
+
+import pytest
+
+from nnnotes.formats import compression
+from nnnotes.pipeline import contract, siteassets
+from nnnotes.pipeline.contract import Input
+from nnnotes.pipeline.depot import Depot
+from nnnotes.pipeline.stages import Env, describe, execute
+
+
+def task_for(depot, data, ext="json", params=None, name="first"):
+    inp = Input("source", depot.put(data), len(data), name, ({"kind": "depot"},))
+    subject = siteassets.subject(inp, ext)
+    env = Env(depot, {"site_assets": {subject: inp}})
+    return describe(siteassets.AssetStage(), subject, params, env)
+
+
+def run(depot, data, ext="json", params=None):
+    task = task_for(depot, data, ext, params)
+    execute(task, depot, {siteassets.ASSET: siteassets.AssetStage()})
+    result = depot.result(task.key)
+    return task, result, siteassets.entry(result, depot)
+
+
+def restored(depot, result, entry):
+    by_path = {p["path"]: p for p in siteassets.placements(result)}
+
+    def read(path, size, stored=None):
+        p = by_path[path]
+        raw = depot.read(p["sha256"])
+        assert len(raw) == p["size"] == (size if stored is None else stored)
+        data = compression.decode(path, raw)
+        assert len(data) == size
+        assert contract.sha256(data) == path.split("/")[1].split(".")[0]
+        return data
+
+    if "parts" not in entry:
+        return read(entry["asset"], entry["size"], entry.get("stored"))
+    body = b"{" + b",".join(json.dumps(k).encode() + b":" + read(*p) for k, *p in entry["parts"]) + b"}"
+    assert len(body) == entry["size"]
+    return body
+
+
+@pytest.mark.parametrize("encoding", ["none", "gzip", "br"])
+def test_assets_reconstruct_projected_json_from_depot_bytes(tmp_path, encoding):
+    depot = Depot(tmp_path)
+    doc = {"title": "text " * 1000, "nested": {"value": 9, "note": "label"}, "signed": -12.5}
+    _, result, entry = run(
+        depot, json.dumps(doc, indent=2).encode(), params={"encoding": encoding, "omit": [["nested", "note"]]}
+    )
+    expected = {**doc, "nested": {"value": 9}}
+    assert json.loads(restored(depot, result, entry)) == expected
+    assert ("stored" in entry) == (encoding != "none")
+    assert all(p["path"].startswith("assets/") for p in siteassets.placements(result))
+
+
+def test_split_parts_share_content_and_reconstruct_exact_text(tmp_path):
+    depot = Depot(tmp_path)
+    data = b'{"a": [1,2,3], "b": [1,2,3], "c": {"x":true}}'
+    _, result, entry = run(depot, data, params={"split": 10})
+    assert len(entry["parts"]) == 3
+    assert entry["parts"][0][1] == entry["parts"][1][1]
+    assert len(siteassets.placements(result)) == 2
+    assert restored(depot, result, entry) == b'{"a":[1,2,3],"b":[1,2,3],"c":{"x":true}}'
+
+
+@pytest.mark.parametrize("data", [b'{"a\\"b":1,"c":2}', b"[1,2,3]", b'{"only":123456}', b"{}"])
+def test_json_without_splittable_top_level_keys_stays_whole(tmp_path, data):
+    depot = Depot(tmp_path)
+    _, result, entry = run(depot, data, params={"split": 1})
+    assert "parts" not in entry
+    assert json.loads(restored(depot, result, entry)) == json.loads(data)
+
+
+def test_binary_and_small_files_keep_original_bytes(tmp_path):
+    depot = Depot(tmp_path)
+    for data, ext in [(bytes(range(256)), "png"), (b"a", "bin")]:
+        _, result, entry = run(depot, data, ext)
+        assert "stored" not in entry
+        assert restored(depot, result, entry) == data
+
+
+def test_task_keys_follow_content_transform_and_encoder_identity(tmp_path, monkeypatch):
+    depot = Depot(tmp_path)
+    data = b'{"x":1}'
+    first = task_for(depot, data)
+    assert first.key == task_for(depot, data, name="second").key
+    assert first.key != task_for(depot, b'{"x":2}').key
+    assert first.key != task_for(depot, data, params={"encoding": "none"}).key
+    assert first.key != task_for(depot, data, params={"split": 0}).key
+    assert first.key != task_for(depot, data, params={"omit": [["x"]]}).key
+    monkeypatch.setattr(siteassets.zlib, "ZLIB_RUNTIME_VERSION", "different")
+    assert first.key != task_for(depot, data).key
+    with pytest.raises(contract.IncompatibleTask, match="encoder"):
+        execute(first, depot, {siteassets.ASSET: siteassets.AssetStage()})
+
+
+def test_task_replay_reuses_result_and_forced_execution_is_identical(tmp_path):
+    depot = Depot(tmp_path)
+    task, result, _ = run(depot, b'{"a":1}')
+    registry = {siteassets.ASSET: siteassets.AssetStage()}
+    assert execute(task, depot, registry).status == "hit"
+    assert execute(task, depot, registry, force=True).status == "ran"
+    assert depot.result(task.key) == result
+
+
+def test_site_assets_are_available_to_the_task_runner():
+    from nnnotes.cli.assets import installed
+
+    stages = installed([siteassets.ASSET])
+    assert isinstance(stages.stages[siteassets.ASSET], siteassets.AssetStage)
+    assert not stages.missing
+
+
+def test_site_asset_task_and_result_obey_shared_schemas(tmp_path):
+    from tests.pipeline.test_contract import validators
+
+    depot = Depot(tmp_path)
+    task, result, _ = run(depot, b'{"a":"' + b"x" * 1024 + b'"}')
+    schemas = validators()
+    for name, doc in [
+        (contract.TASK, task.to_json()),
+        (contract.RESULT, result),
+        *((contract.ARTIFACT, a) for a in result["artifacts"]),
+    ]:
+        assert not list(schemas[name].iter_errors(doc))
+
+
+@pytest.mark.parametrize(
+    "params", [{"encoding": "zip"}, {"split": -1}, {"split": True}, {"omit": ["x"]}, {"omit": [[]]}]
+)
+def test_asset_parameters_are_validated_before_execution(params):
+    with pytest.raises(ValueError):
+        siteassets.AssetStage().normalize(params)

tests/site/test_compress.py

@@ -6,7 +6,7 @@
 
 import pytest
 
-from nnnotes.site import compress
+from nnnotes.formats import compression as compress
 
 TEXT = json.dumps({str(i): [i, i * i, f"v{i % 7}"] for i in range(500)}).encode()
 NOISE = random.Random(7).randbytes(4096)