diff --git a/docs/stages.md b/docs/stages.md
index c88b33b..cf8be03 100644
--- a/docs/stages.md
+++ b/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 `.`;
+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.
diff --git a/src/nnnotes/cli/assets.py b/src/nnnotes/cli/assets.py
index a6c639a..884e26c 100644
--- a/src/nnnotes/cli/assets.py
+++ b/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),
)
diff --git a/src/nnnotes/cli/site.py b/src/nnnotes/cli/site.py
index 1849734..c0f0557 100644
--- a/src/nnnotes/cli/site.py
+++ b/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 (
diff --git a/src/nnnotes/site/compress.py b/src/nnnotes/formats/compression.py
similarity index 100%
rename from src/nnnotes/site/compress.py
rename to src/nnnotes/formats/compression.py
diff --git a/src/nnnotes/pipeline/siteassets.py b/src/nnnotes/pipeline/siteassets.py
new file mode 100644
index 0000000..2f325fa
--- /dev/null
+++ b/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"
+ ]
diff --git a/src/nnnotes/site/sitefiles.py b/src/nnnotes/site/sitefiles.py
index 9f47ec5..0fd4d12 100644
--- a/src/nnnotes/site/sitefiles.py
+++ b/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)
diff --git a/src/nnnotes/site/siteindex.py b/src/nnnotes/site/siteindex.py
index df2d611..a6f9a2f 100644
--- a/src/nnnotes/site/siteindex.py
+++ b/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,
diff --git a/src/nnnotes/site/web.py b/src/nnnotes/site/web.py
index cb78532..6b750f5 100644
--- a/src/nnnotes/site/web.py
+++ b/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,
diff --git a/tests/pipeline/test_siteassets.py b/tests/pipeline/test_siteassets.py
new file mode 100644
index 0000000..98f17ec
--- /dev/null
+++ b/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)
diff --git a/tests/site/test_compress.py b/tests/site/test_compress.py
index c702bb3..b1789e2 100644
--- a/tests/site/test_compress.py
+++ b/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)
diff --git a/tests/site/test_web.py b/tests/site/test_web.py
index ac12eb9..cedda82 100644
--- a/tests/site/test_web.py
+++ b/tests/site/test_web.py
@@ -4,7 +4,8 @@
import pytest
-from nnnotes.site import compress, sitefiles, siteindex, web
+from nnnotes.formats import compression as compress
+from nnnotes.site import sitefiles, siteindex, web
from nnnotes.source.config import ConfigError
DOC = sitefiles._minify({"rows": [{"id": i, "name": f"row {i}", "tags": ["a", "b"]} for i in range(200)]}).encode()