diff --git a/dockerfiles/hermes-hux-foundation/hux/artifacts.py b/dockerfiles/hermes-hux-foundation/hux/artifacts.py
new file mode 100644
index 00000000..6eea3776
--- /dev/null
+++ b/dockerfiles/hermes-hux-foundation/hux/artifacts.py
@@ -0,0 +1,393 @@
+"""HUX-04 artifact workspace: typed, versioned, diffable outputs.
+
+Every artifact is one ``hux.artifact.v1`` document under the caller's tenant
+subtree. Content lives in the content-addressed blob store; the document only
+carries ``content_ref`` hashes, so a version can never be rewritten once it
+has been appended (SO-30). Blobs are served inside a JSON envelope with
+``nosniff`` and never as an executable response type (SO-32). Every
+referenced id is resolved under the caller's own subtree (SO-33); a foreign
+or unknown id is ``404`` so ids cannot be probed. Sharing is out of scope for
+this increment: ``access.mode`` is always ``owner`` (SO-34).
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+from typing import Any
+
+from hux import diffs
+from hux.contracts import load_all, validate_record
+from hux.errors import Conflict, Invalid, NotFound, TooLarge
+from hux.http import Request, Response, Router, page
+from hux.identity import Identity
+from hux.store import TenantStore, check_id, new_id, now_iso
+
+FAMILY = "artifacts"
+MAX_VERSION_BYTES = 25 * 1024 * 1024
+MAX_VERSIONS = 200
+MAX_ARTIFACTS = 2000
+PAGE_SIZE = 50
+ARTIFACT_TYPES = ("markdown", "code", "html", "svg", "image", "json", "csv", "document", "audio")
+SENSITIVITIES = ("public", "personal", "sensitive", "restricted")
+_SCHEMAS: dict[str, dict[str, Any]] = {}
+
+
+def _schemas() -> dict[str, dict[str, Any]]:
+    if not _SCHEMAS:
+        _SCHEMAS.update(load_all())
+    return _SCHEMAS
+
+
+# -- helpers shared with other lanes -------------------------------------------
+
+def artifact_exists(store: TenantStore, artifact_id: str) -> bool:
+    """True when ``artifact_id`` is a well-formed id owned by the store's tenant."""
+    try:
+        return store.exists(FAMILY, artifact_id)
+    except Invalid:
+        return False
+
+
+def artifact_titles(store: TenantStore, ids: list[str]) -> dict[str, str]:
+    """Map each resolvable artifact id to its title; unknown ids are omitted."""
+    titles: dict[str, str] = {}
+    for artifact_id in ids:
+        if artifact_exists(store, artifact_id):
+            titles[artifact_id] = store.get(FAMILY, artifact_id)["title"]
+    return titles
+
+
+def emit_event(store: TenantStore, identity: Identity, artifact: dict[str, Any], kind: str, summary: str, detail: dict[str, Any]) -> None:
+    """Record an activity event through the events lane when it is present.
+
+    Imported lazily so module import order never matters; a missing events
+    module (earlier build) is not an error because the artifact write itself
+    is the source of truth.
+    """
+    conversation_id = artifact.get("conversation_id")
+    if conversation_id is None:
+        return
+    try:
+        from hux.events import emit
+    except ModuleNotFoundError:
+        return
+    version = artifact["current_version"]
+    evidence = [{"kind": "artifact_version", "id": f"{artifact['id']}@{version}", "hash": artifact["versions"][-1]["content_ref"]["hash"]}]
+    emit(store, identity, conversation_id, kind, summary, detail=detail, evidence=evidence, sensitivity=artifact["sensitivity"])
+
+
+def replay(store: TenantStore, family: str, key: str) -> str | None:
+    """Record id previously created under an Idempotency-Key, if any."""
+    if not key:
+        return None
+    for row in store.read(family, "idempotency"):
+        if row.get("key") == key:
+            return row["id"]
+    return None
+
+
+def remember(store: TenantStore, family: str, key: str, record_id: str) -> None:
+    """Persist an Idempotency-Key to record id mapping."""
+    if key:
+        store.append(family, "idempotency", {"key": key, "id": record_id, "at": now_iso()})
+
+
+# -- record shaping ------------------------------------------------------------
+
+def _actor(identity: Identity) -> dict[str, str]:
+    if identity.trust == "worker":
+        return {"type": "system", "id": "worker"}
+    return {"type": "user", "id": identity.subject}
+
+
+def _body(request: Request) -> dict[str, Any]:
+    if not isinstance(request.body, dict):
+        raise Invalid("body must be a JSON object")
+    return request.body
+
+
+def _string(body: dict[str, Any], key: str, limit: int, required: bool = True) -> str | None:
+    value = body.get(key)
+    if value is None:
+        if required:
+            raise Invalid(f"{key} is required")
+        return None
+    if not isinstance(value, str) or not value.strip() or len(value) > limit:
+        raise Invalid(f"{key} must be a non-empty string of at most {limit} characters")
+    return value
+
+
+def _content(body: dict[str, Any]) -> tuple[bytes, str]:
+    """Decode the submitted content and return ``(bytes, mime)``."""
+    text, encoded = body.get("content"), body.get("content_base64")
+    if isinstance(text, str) and encoded is None:
+        data = text.encode("utf-8")
+    elif isinstance(encoded, str) and text is None:
+        try:
+            data = base64.b64decode(encoded, validate=True)
+        except (ValueError, TypeError) as error:
+            raise Invalid("content_base64 is not valid base64") from error
+    else:
+        raise Invalid("exactly one of content (utf-8 text) or content_base64 is required")
+    if len(data) > MAX_VERSION_BYTES:
+        raise TooLarge("version exceeds 25 MiB")
+    mime = _string(body, "mime", 120, required=False) or "application/octet-stream"
+    return data, mime
+
+
+def _store_content(store: TenantStore, body: dict[str, Any]) -> dict[str, Any]:
+    """Hash, verify against any client-supplied hash, store the blob, return ``content_ref``."""
+    data, mime = _content(body)
+    digest = hashlib.sha256(data).hexdigest()
+    claimed = body.get("hash")
+    if claimed is not None and claimed != f"sha256:{digest}":
+        raise Invalid("hash does not match the submitted content")
+    store.put_blob(digest, data)
+    return {"hash": f"sha256:{digest}", "bytes": len(data), "mime": mime}
+
+
+def _optional_id(body: dict[str, Any], key: str) -> str | None:
+    value = body.get(key)
+    return None if value is None else check_id(value)
+
+
+def _load_owned(request: Request) -> dict[str, Any]:
+    """The artifact named in the path; unknown, malformed or foreign is 404."""
+    try:
+        artifact = request.store.get(FAMILY, request.params["id"])
+    except Invalid as error:
+        raise NotFound("artifact not found") from error
+    if artifact.get("owner") != request.identity.subject:
+        raise NotFound("artifact not found")
+    return artifact
+
+
+def _version(artifact: dict[str, Any], number: Any) -> dict[str, Any]:
+    if not isinstance(number, int) or isinstance(number, bool):
+        raise Invalid("version must be an integer")
+    for entry in artifact["versions"]:
+        if entry["version"] == number:
+            return entry
+    raise NotFound(f"version {number} not found")
+
+
+def _lineage(request: Request, body: dict[str, Any]) -> dict[str, Any] | None:
+    """Resolve ``lineage`` under the caller's subtree; anything else is 404 (SO-33)."""
+    lineage = body.get("lineage")
+    if lineage is None:
+        return None
+    if not isinstance(lineage, dict):
+        raise Invalid("lineage must be an object")
+    try:
+        parent = request.store.get(FAMILY, check_id(lineage.get("artifact_id")))
+    except (Invalid, NotFound) as error:
+        raise NotFound("lineage artifact not found") from error
+    if parent.get("owner") != request.identity.subject:
+        raise NotFound("lineage artifact not found")
+    version = _version(parent, lineage.get("version"))
+    return {"artifact_id": parent["id"], "version": version["version"]}
+
+
+def append_version(store: TenantStore, artifact: dict[str, Any], entry: dict[str, Any], expected_revision: int | None) -> dict[str, Any]:
+    """Append an immutable version entry and persist the document.
+
+    The version number must be exactly ``current_version + 1``; any attempt to
+    write an existing number is a Conflict so history is never rewritten.
+    """
+    if any(v["version"] == entry["version"] for v in artifact["versions"]):
+        raise Conflict(f"version {entry['version']} already exists")
+    if entry["version"] != artifact["current_version"] + 1:
+        raise Conflict("versions are appended in order")
+    if len(artifact["versions"]) >= MAX_VERSIONS:
+        raise TooLarge("artifact has reached 200 versions")
+    updated = {**artifact, "versions": [*artifact["versions"], entry], "current_version": entry["version"], "updated_at": now_iso()}
+    _check(updated)
+    return store.put(FAMILY, updated, expected_revision=expected_revision)
+
+
+def _check(artifact: dict[str, Any]) -> None:
+    problems = validate_record(artifact, _schemas())
+    if problems:
+        raise Invalid("artifact does not satisfy hux.artifact.v1", problems)
+
+
+# -- handlers ------------------------------------------------------------------
+
+def create(request: Request) -> Response:
+    """``POST /hux/v1/artifacts``: create an artifact with version 1."""
+    body = _body(request)
+    key = request.idempotency_key()
+    with request.store.lock(FAMILY):
+        existing = replay(request.store, FAMILY, key)
+        if existing is not None:
+            request.audit("artifacts.create", existing, reason="idempotent_replay")
+            return Response(200, request.store.get(FAMILY, existing), {"HUX-Replayed": "true"})
+        if request.store.count(FAMILY) >= MAX_ARTIFACTS:
+            raise TooLarge("tenant has reached 2000 artifacts")
+        artifact_type = _string(body, "type", 40)
+        if artifact_type not in ARTIFACT_TYPES:
+            raise Invalid("unknown artifact type")
+        sensitivity = body.get("sensitivity", "personal")
+        if sensitivity not in SENSITIVITIES:
+            raise Invalid("unknown sensitivity")
+        stamp = now_iso()
+        version: dict[str, Any] = {"version": 1, "created_at": stamp, "created_by": _actor(request.identity), "content_ref": _store_content(request.store, body)}
+        for field, limit in (("message_id", 120), ("note", 200)):
+            if _string(body, field, limit, required=False) is not None:
+                version[field] = body[field]
+        lineage = _lineage(request, body)
+        if lineage is not None:
+            version["lineage"] = lineage
+        artifact: dict[str, Any] = {
+            "schema": "hux.artifact.v1",
+            "id": new_id("art"),
+            "owner": request.identity.subject,
+            "type": artifact_type,
+            "title": _string(body, "title", 200),
+            "current_version": 1,
+            "versions": [version],
+            "sensitivity": sensitivity,
+            "created_at": stamp,
+            "updated_at": stamp,
+            "revision": 1,
+            "access": {"mode": "owner"},
+        }
+        for field in ("conversation_id", "project_id"):
+            if _optional_id(body, field) is not None:
+                artifact[field] = body[field]
+        if _string(body, "language", 40, required=False) is not None:
+            artifact["language"] = body["language"]
+        _check(artifact)
+        stored = request.store.put(FAMILY, artifact)
+        remember(request.store, FAMILY, key, stored["id"])
+    request.audit("artifacts.create", stored["id"])
+    emit_event(request.store, request.identity, stored, "artifact.created", f"Created {stored['type']} artifact", {"artifact_id": stored["id"], "version": 1})
+    return Response(201, stored, {"ETag": str(stored["revision"])})
+
+
+def list_artifacts(request: Request) -> Response:
+    """``GET /hux/v1/artifacts?conversation_id=&project_id=&cursor=``: owned artifacts, oldest first."""
+    filters = {k: request.query[k] for k in ("conversation_id", "project_id") if request.query.get(k)}
+    cursor = request.query.get("cursor", "0")
+    if not cursor.isdigit():
+        raise Invalid("cursor must be a non-negative integer")
+    items = [
+        artifact
+        for artifact in request.store.scan(FAMILY)
+        if artifact.get("owner") == request.identity.subject and all(artifact.get(k) == v for k, v in filters.items())
+    ]
+    start = int(cursor)
+    window = items[start : start + PAGE_SIZE]
+    request.audit("artifacts.list", "artifacts")
+    return page(window, str(start + PAGE_SIZE) if len(items) > start + PAGE_SIZE else None)
+
+
+def get(request: Request) -> Response:
+    """``GET /hux/v1/artifacts/{id}``: one artifact document."""
+    artifact = _load_owned(request)
+    request.audit("artifacts.get", artifact["id"])
+    return Response(200, artifact, {"ETag": str(artifact["revision"])})
+
+
+def add_version(request: Request) -> Response:
+    """``POST /hux/v1/artifacts/{id}/versions``: append a new immutable version (If-Match required)."""
+    body = _body(request)
+    expected = request.if_match()
+    if expected is None:
+        raise Invalid("If-Match is required to append a version")
+    key = request.idempotency_key()
+    with request.store.lock(FAMILY):
+        artifact = _load_owned(request)
+        existing = replay(request.store, FAMILY, key)
+        if existing is not None and existing.split("@")[0] == artifact["id"]:
+            request.audit("artifacts.version", existing, reason="idempotent_replay")
+            return Response(200, artifact, {"HUX-Replayed": "true", "ETag": str(artifact["revision"])})
+        if expected != artifact["revision"]:
+            raise Conflict(f"revision {expected} does not match current revision {artifact['revision']}", [str(artifact["revision"])])
+        number = artifact["current_version"] + 1
+        entry: dict[str, Any] = {"version": number, "created_at": now_iso(), "created_by": _actor(request.identity), "content_ref": _store_content(request.store, body), "diff_from": artifact["current_version"]}
+        if body.get("diff_from") is not None:
+            entry["diff_from"] = _version(artifact, body["diff_from"])["version"]
+        for field, limit in (("message_id", 120), ("note", 200)):
+            if _string(body, field, limit, required=False) is not None:
+                entry[field] = body[field]
+        lineage = _lineage(request, body)
+        if lineage is not None:
+            entry["lineage"] = lineage
+        stored = append_version(request.store, artifact, entry, expected)
+        remember(request.store, FAMILY, key, f"{stored['id']}@{number}")
+    request.audit("artifacts.version", f"{stored['id']}@{number}")
+    emit_event(request.store, request.identity, stored, "artifact.version", f"New version {number}", {"artifact_id": stored["id"], "version": number})
+    return Response(201, stored, {"ETag": str(stored["revision"])})
+
+
+def _version_number(request: Request) -> int:
+    raw = request.params["n"]
+    if not raw.isdigit() or int(raw) < 1:
+        raise Invalid("version must be a positive integer")
+    return int(raw)
+
+
+def get_version(request: Request) -> Response:
+    """``GET /hux/v1/artifacts/{id}/versions/{n}``: version metadata plus content."""
+    artifact = _load_owned(request)
+    entry = _version(artifact, _version_number(request))
+    data = request.store.get_blob(entry["content_ref"]["hash"].split(":", 1)[1])
+    text = diffs.decode_text(data) if diffs.is_text_type(artifact["type"]) else None
+    body: dict[str, Any] = {"artifact_id": artifact["id"], "type": artifact["type"], "version": entry}
+    if text is None:
+        body["content_base64"] = base64.b64encode(data).decode("ascii")
+    else:
+        body["content"] = text
+    request.audit("artifacts.get_version", f"{artifact['id']}@{entry['version']}")
+    headers = {"X-Content-Type-Options": "nosniff", "Content-Disposition": "attachment", "ETag": str(artifact["revision"])}
+    return Response(200, body, headers)
+
+
+def diff(request: Request) -> Response:
+    """``GET /hux/v1/artifacts/{id}/versions/{n}/diff?from=``: unified diff for text, sizes and hashes otherwise."""
+    artifact = _load_owned(request)
+    to_entry = _version(artifact, _version_number(request))
+    raw_from = request.query.get("from", "")
+    if raw_from and not raw_from.isdigit():
+        raise Invalid("from must be a version number")
+    from_number = int(raw_from) if raw_from else to_entry.get("diff_from", to_entry["version"])
+    from_entry = _version(artifact, from_number)
+    older = request.store.get_blob(from_entry["content_ref"]["hash"].split(":", 1)[1])
+    newer = request.store.get_blob(to_entry["content_ref"]["hash"].split(":", 1)[1])
+    request.audit("artifacts.diff", f"{artifact['id']}@{from_number}..{to_entry['version']}")
+    return Response(200, diffs.unified(artifact["type"], older, newer, from_number, to_entry["version"]), {"X-Content-Type-Options": "nosniff"})
+
+
+def promote(request: Request) -> Response:
+    """``POST /hux/v1/artifacts/{id}/promote``: mark the current version as the project's copy."""
+    body = _body(request)
+    project_id = check_id(body.get("project_id"))
+    expected = request.if_match()
+    with request.store.lock(FAMILY):
+        artifact = _load_owned(request)
+        if expected is not None and expected != artifact["revision"]:
+            raise Conflict(f"revision {expected} does not match current revision {artifact['revision']}", [str(artifact["revision"])])
+        version = artifact["current_version"]
+        if body.get("version") is not None:
+            version = _version(artifact, body["version"])["version"]
+        stamp = now_iso()
+        updated = {**artifact, "project_id": project_id, "promotion": {"project_id": project_id, "version": version, "at": stamp}, "updated_at": stamp}
+        _check(updated)
+        stored = request.store.put(FAMILY, updated, expected_revision=artifact["revision"])
+    request.audit("artifacts.promote", f"{stored['id']}@{version}", reason="" if expected is not None else "unconditional_write")
+    emit_event(request.store, request.identity, stored, "artifact.promoted", f"Promoted version {version} to project", {"artifact_id": stored["id"], "version": version, "project_id": project_id})
+    return Response(200, stored, {"ETag": str(stored["revision"])})
+
+
+def register(router: Router) -> None:
+    """Attach HUX-04 routes."""
+    card = "HUX-04"
+    router.add("POST", "/hux/v1/artifacts", card, "artifacts.create", create)
+    router.add("GET", "/hux/v1/artifacts", card, "artifacts.list", list_artifacts)
+    router.add("GET", "/hux/v1/artifacts/{id}", card, "artifacts.get", get)
+    router.add("POST", "/hux/v1/artifacts/{id}/versions", card, "artifacts.version", add_version)
+    router.add("GET", "/hux/v1/artifacts/{id}/versions/{n}", card, "artifacts.get_version", get_version)
+    router.add("GET", "/hux/v1/artifacts/{id}/versions/{n}/diff", card, "artifacts.diff", diff)
+    router.add("POST", "/hux/v1/artifacts/{id}/promote", card, "artifacts.promote", promote)
diff --git a/dockerfiles/hermes-hux-foundation/hux/diffs.py b/dockerfiles/hermes-hux-foundation/hux/diffs.py
new file mode 100644
index 00000000..91dd719c
--- /dev/null
+++ b/dockerfiles/hermes-hux-foundation/hux/diffs.py
@@ -0,0 +1,50 @@
+"""Version-to-version comparison for artifacts (HUX-04).
+
+Text artifacts get a unified diff from ``difflib``; anything that is not
+valid UTF-8, or whose artifact type is binary, is described by sizes and
+hashes only so the UI never tries to render bytes as text.
+"""
+
+from __future__ import annotations
+
+import difflib
+import hashlib
+from typing import Any
+
+TEXT_TYPES = frozenset({"markdown", "code", "html", "svg", "json", "csv"})
+
+
+def is_text_type(artifact_type: str) -> bool:
+    """True for artifact types whose content is served as UTF-8 text."""
+    return artifact_type in TEXT_TYPES
+
+
+def decode_text(data: bytes) -> str | None:
+    """UTF-8 decode, or None when the bytes are not text."""
+    try:
+        return data.decode("utf-8")
+    except UnicodeDecodeError:
+        return None
+
+
+def unified(artifact_type: str, older: bytes, newer: bytes, from_n: int, to_n: int) -> dict[str, Any]:
+    """Diff record ``{"from", "to", "unified"|"binary"}`` for two versions."""
+    record: dict[str, Any] = {"from": from_n, "to": to_n}
+    old_text = decode_text(older) if is_text_type(artifact_type) else None
+    new_text = decode_text(newer) if is_text_type(artifact_type) else None
+    if old_text is None or new_text is None:
+        record["binary"] = {
+            "from_bytes": len(older),
+            "to_bytes": len(newer),
+            "from_hash": "sha256:" + hashlib.sha256(older).hexdigest(),
+            "to_hash": "sha256:" + hashlib.sha256(newer).hexdigest(),
+        }
+        return record
+    lines = difflib.unified_diff(
+        old_text.splitlines(keepends=True),
+        new_text.splitlines(keepends=True),
+        fromfile=f"v{from_n}",
+        tofile=f"v{to_n}",
+    )
+    record["unified"] = "".join(lines)
+    return record
diff --git a/dockerfiles/hermes-hux-foundation/hux/research.py b/dockerfiles/hermes-hux-foundation/hux/research.py
new file mode 100644
index 00000000..90b07a18
--- /dev/null
+++ b/dockerfiles/hermes-hux-foundation/hux/research.py
@@ -0,0 +1,378 @@
+"""HUX-08 research model: sources, passages, citations and notebooks.
+
+A source says where evidence came from, a passage is the exact excerpt, a
+citation ties a claim in a message to passages with a support verdict, and a
+notebook collects them per research question. The service never fetches a
+``uri`` (SO-19, SO-29): URIs are stored metadata under an allowlisted scheme.
+Every referenced id resolves under the caller's own subtree (SO-33); a foreign
+or unknown id is ``404`` so ids cannot be probed. Duplicates are merged on a
+server-computed ``dedupe_key`` and the original record is returned.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Any
+from urllib.parse import urlsplit, urlunsplit
+
+from hux.artifacts import remember, replay
+from hux.contracts import load_all, validate_record
+from hux.errors import Conflict, Invalid, NotFound, TooLarge
+from hux.http import Request, Response, Router, page
+from hux.identity import Identity
+from hux.store import TenantStore, check_id, new_id, now_iso
+
+FAMILY = "research"
+SOURCES, PASSAGES, CITATIONS, NOTEBOOKS = "sources", "passages", "citations", "notebooks"
+SCHEMES = frozenset({"http", "https", "artifact", "memory", "file"})
+SOURCE_KINDS = ("web", "document", "artifact", "memory", "tool_output", "dataset")
+CLASSIFICATIONS = ("primary", "secondary", "unknown")
+SUPPORT = ("supports", "partially_supports", "contradicts", "unverified")
+NOTEBOOK_TRANSITIONS = {"open": frozenset({"answered", "abandoned"}), "answered": frozenset(), "abandoned": frozenset()}
+MAX_SOURCES, MAX_PASSAGES = 10000, 20000
+_SCHEMAS: dict[str, dict[str, Any]] = {}
+
+
+def _schemas() -> dict[str, dict[str, Any]]:
+    if not _SCHEMAS:
+        _SCHEMAS.update(load_all())
+    return _SCHEMAS
+
+
+def _check(record: dict[str, Any]) -> None:
+    problems = validate_record(record, _schemas())
+    if problems:
+        raise Invalid(f"{record.get('schema')} record is not valid", problems)
+
+
+def _body(request: Request) -> dict[str, Any]:
+    if not isinstance(request.body, dict):
+        raise Invalid("body must be a JSON object")
+    return request.body
+
+
+def _string(body: dict[str, Any], key: str, limit: int, required: bool = True) -> str | None:
+    value = body.get(key)
+    if value is None:
+        if required:
+            raise Invalid(f"{key} is required")
+        return None
+    if not isinstance(value, str) or not value.strip() or len(value) > limit:
+        raise Invalid(f"{key} must be a non-empty string of at most {limit} characters")
+    return value
+
+
+def _sha(*parts: str) -> str:
+    return "sha256:" + hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest()
+
+
+def _lookup(store: TenantStore, ledger: str, key: str) -> str | None:
+    for row in store.read(FAMILY, ledger):
+        if row.get("key") == key:
+            return row["id"]
+    return None
+
+
+def _resolve(store: TenantStore, family: str, record_id: Any) -> dict[str, Any]:
+    """Load a record under the caller's subtree; malformed or missing is 404 (SO-33).
+
+    Sources, passages and citations are immutable, so the store's ``revision``
+    bookkeeping is stripped before they are served; notebooks keep theirs.
+    """
+    try:
+        record = store.get(family, check_id(record_id))
+    except (Invalid, NotFound) as error:
+        raise NotFound(f"{family[:-1]} not found") from error
+    return record if family == NOTEBOOKS else {k: v for k, v in record.items() if k != "revision"}
+
+
+def _id_list(body: dict[str, Any], key: str, maximum: int = 32) -> list[str]:
+    values = body.get(key, [])
+    if not isinstance(values, list) or not all(isinstance(v, str) for v in values):
+        raise Invalid(f"{key} must be a list of ids")
+    unique = list(dict.fromkeys(values))
+    if len(unique) > maximum:
+        raise TooLarge(f"{key} exceeds {maximum} ids")
+    return unique
+
+
+def _emit(store: TenantStore, identity: Identity, conversation_id: str | None, kind: str, summary: str, detail: dict[str, Any], evidence: list[dict[str, Any]]) -> None:
+    if conversation_id is None:
+        return
+    try:
+        from hux.events import emit
+    except ModuleNotFoundError:
+        return
+    emit(store, identity, conversation_id, kind, summary, detail=detail, evidence=evidence)
+
+
+# -- sources ---------------------------------------------------------------------
+
+def normalise_uri(uri: str) -> str:
+    """Canonical form used for dedupe: lowercase scheme and host, no fragment, no trailing slash."""
+    parts = urlsplit(uri.strip())
+    if parts.scheme.lower() not in SCHEMES:
+        raise Invalid("uri scheme must be one of http, https, artifact, memory, file")
+    path = parts.path.rstrip("/") or ("/" if parts.netloc else "")
+    return urlunsplit((parts.scheme.lower(), parts.netloc.lower(), path, parts.query, ""))
+
+
+def create_source(request: Request) -> Response:
+    """``POST /hux/v1/sources``: record where evidence came from; duplicates return the original."""
+    body = _body(request)
+    key = request.idempotency_key()
+    kind = _string(body, "kind", 40)
+    if kind not in SOURCE_KINDS:
+        raise Invalid("unknown source kind")
+    classification = body.get("classification", "unknown")
+    if classification not in CLASSIFICATIONS:
+        raise Invalid("unknown classification")
+    title = _string(body, "title", 300)
+    uri = _string(body, "uri", 2000, required=False)
+    dedupe = _sha("uri", normalise_uri(uri)) if uri is not None else _sha("title", kind, title.strip().lower())
+    stamp = now_iso()
+    with request.store.lock(FAMILY):
+        existing = replay(request.store, FAMILY, key) or _lookup(request.store, "source_keys", dedupe)
+        if existing is not None:
+            request.audit("research.source_create", existing, reason="deduplicated")
+            return Response(200, _resolve(request.store, SOURCES, existing), {"HUX-Replayed": "true"})
+        if request.store.count(SOURCES) >= MAX_SOURCES:
+            raise TooLarge("tenant has reached 10000 sources")
+        record: dict[str, Any] = {
+            "schema": "hux.source.v1", "id": new_id("src"), "kind": kind, "title": title,
+            "classification": classification, "retrieved_at": stamp, "dedupe_key": dedupe,
+            "provenance": {"surface": request.identity.surface, "actor": _actor(request.identity), "recorded_at": stamp},
+        }
+        if uri is not None:
+            record["uri"] = uri
+        for field, limit in (("publisher", 200), ("published_at", 40), ("content_hash", 71)):
+            if _string(body, field, limit, required=False) is not None:
+                record[field] = body[field]
+        conversation_id = body.get("conversation_id")
+        if conversation_id is not None:
+            record["provenance"]["conversation_id"] = check_id(conversation_id)
+        _check(record)
+        stored = record | {}
+        request.store.put(SOURCES, record)
+        request.store.append(FAMILY, "source_keys", {"key": dedupe, "id": stored["id"]})
+        remember(request.store, FAMILY, key, stored["id"])
+    request.audit("research.source_create", stored["id"])
+    return Response(201, stored)
+
+
+def _actor(identity: Identity) -> dict[str, str]:
+    if identity.trust == "worker":
+        return {"type": "system", "id": "worker"}
+    return {"type": "user", "id": identity.subject}
+
+
+def get_source(request: Request) -> Response:
+    """``GET /hux/v1/sources/{id}``: one source record."""
+    record = _resolve(request.store, SOURCES, request.params["id"])
+    request.audit("research.source_get", record["id"])
+    return Response(200, record)
+
+
+# -- passages --------------------------------------------------------------------
+
+def create_passage(request: Request) -> Response:
+    """``POST /hux/v1/passages``: an exact excerpt of a source; hash and dedupe key are server-computed."""
+    body = _body(request)
+    key = request.idempotency_key()
+    source = _resolve(request.store, SOURCES, body.get("source_id"))
+    text = _string(body, "text", 4000)
+    text_hash = "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest()
+    dedupe = _sha("passage", source["id"], text_hash)
+    with request.store.lock(FAMILY):
+        existing = replay(request.store, FAMILY, key) or _lookup(request.store, "passage_keys", dedupe)
+        if existing is not None:
+            request.audit("research.passage_create", existing, reason="deduplicated")
+            return Response(200, _resolve(request.store, PASSAGES, existing), {"HUX-Replayed": "true"})
+        if request.store.count(PASSAGES) >= MAX_PASSAGES:
+            raise TooLarge("tenant has reached 20000 passages")
+        record: dict[str, Any] = {"schema": "hux.passage.v1", "id": new_id("psg"), "source_id": source["id"], "text": text, "hash": text_hash, "dedupe_key": dedupe}
+        if body.get("locator") is not None:
+            record["locator"] = body["locator"]
+        _check(record)
+        stored = record | {}
+        request.store.put(PASSAGES, record)
+        request.store.append(FAMILY, "passage_keys", {"key": dedupe, "id": stored["id"]})
+        remember(request.store, FAMILY, key, stored["id"])
+    request.audit("research.passage_create", stored["id"])
+    return Response(201, stored)
+
+
+# -- citations -------------------------------------------------------------------
+
+def _message_ledger(message_id: str) -> str:
+    return "msg_" + hashlib.sha256(message_id.encode("utf-8")).hexdigest()[:40]
+
+
+def attach_citation(request: Request) -> Response:
+    """``POST /hux/v1/messages/{id}/citations``: tie a claim to passages with a support verdict."""
+    body = _body(request)
+    key = request.idempotency_key()
+    message_id = request.params["id"]
+    if len(message_id) > 120:
+        raise Invalid("message id is too long")
+    claim = _string(body, "claim", 1000)
+    support = body.get("support", "unverified")
+    if support not in SUPPORT:
+        raise Invalid("unknown support verdict")
+    passage_ids = _id_list(body, "passage_ids")
+    if not passage_ids:
+        raise Invalid("at least one passage_id is required")
+    passages = [_resolve(request.store, PASSAGES, pid) for pid in passage_ids]
+    dedupe = _sha("citation", message_id, claim, *sorted(passage_ids))
+    with request.store.lock(FAMILY):
+        existing = replay(request.store, FAMILY, key) or _lookup(request.store, "citation_keys", dedupe)
+        if existing is not None:
+            request.audit("research.citation_attach", existing, reason="deduplicated")
+            return Response(200, _resolve(request.store, CITATIONS, existing), {"HUX-Replayed": "true"})
+        record: dict[str, Any] = {"schema": "hux.citation.v1", "id": new_id("cit"), "message_id": message_id, "claim": claim, "passage_ids": passage_ids, "support": support, "dedupe_key": dedupe}
+        if _string(body, "note", 500, required=False) is not None:
+            record["note"] = body["note"]
+        _check(record)
+        stored = record | {}
+        request.store.put(CITATIONS, record)
+        request.store.append(FAMILY, "citation_keys", {"key": dedupe, "id": stored["id"]})
+        request.store.append(FAMILY, _message_ledger(message_id), {"id": stored["id"]})
+        remember(request.store, FAMILY, key, stored["id"])
+    request.audit("research.citation_attach", stored["id"])
+    evidence = [{"kind": "passage", "id": p["id"], "hash": p["hash"]} for p in passages]
+    conversation_id = body.get("conversation_id")
+    _emit(request.store, request.identity, None if conversation_id is None else check_id(conversation_id), "citation.attached", f"Citation attached ({support})", {"message_id": message_id, "citation_id": stored["id"], "support": support}, evidence)
+    return Response(201, stored)
+
+
+def list_citations(request: Request) -> Response:
+    """``GET /hux/v1/messages/{id}/citations``: citations with passages and sources embedded."""
+    items = []
+    for row in request.store.read(FAMILY, _message_ledger(request.params["id"])):
+        citation = _resolve(request.store, CITATIONS, row["id"])
+        passages = [_resolve(request.store, PASSAGES, pid) for pid in citation["passage_ids"]]
+        sources = {p["source_id"]: _resolve(request.store, SOURCES, p["source_id"]) for p in passages}
+        items.append({"citation": citation, "passages": passages, "sources": list(sources.values())})
+    request.audit("research.citation_list", request.params["id"])
+    return page(items)
+
+
+def validate_citations(store: TenantStore, citation_ids: list[str]) -> list[str]:
+    """Integrity problems for a set of citations: dangling passages, passage/source mismatch, contradictions without a note."""
+    problems: list[str] = []
+    for citation_id in citation_ids:
+        try:
+            citation = _resolve(store, CITATIONS, citation_id)
+        except NotFound:
+            problems.append(f"{citation_id}: citation does not resolve")
+            continue
+        if citation["support"] == "contradicts" and not citation.get("note"):
+            problems.append(f"{citation_id}: contradicts without a note")
+        for passage_id in citation["passage_ids"]:
+            try:
+                passage = _resolve(store, PASSAGES, passage_id)
+            except NotFound:
+                problems.append(f"{citation_id}: passage {passage_id} does not resolve")
+                continue
+            if not store.exists(SOURCES, passage["source_id"]):
+                problems.append(f"{citation_id}: passage {passage_id} names missing source {passage['source_id']}")
+    return problems
+
+
+# -- notebooks -------------------------------------------------------------------
+
+def _text_list(body: dict[str, Any], key: str, current: list[str]) -> list[str]:
+    values = body.get(key)
+    if values is None:
+        return current
+    if not isinstance(values, list) or not all(isinstance(v, str) and v.strip() for v in values):
+        raise Invalid(f"{key} must be a list of non-empty strings")
+    return values
+
+
+def create_notebook(request: Request) -> Response:
+    """``POST /hux/v1/notebooks``: open a research question for a conversation."""
+    body = _body(request)
+    key = request.idempotency_key()
+    with request.store.lock(FAMILY):
+        existing = replay(request.store, FAMILY, key)
+        if existing is not None:
+            request.audit("research.notebook_create", existing, reason="idempotent_replay")
+            return Response(200, request.store.get(NOTEBOOKS, existing), {"HUX-Replayed": "true"})
+        record: dict[str, Any] = {
+            "schema": "hux.research_notebook.v1", "id": new_id("nb"), "conversation_id": check_id(body.get("conversation_id")),
+            "question": _string(body, "question", 1000), "status": "open", "source_ids": [], "passage_ids": [], "citation_ids": [],
+            "assumptions": _text_list(body, "assumptions", []), "unresolved_questions": _text_list(body, "unresolved_questions", []),
+            "updated_at": now_iso(), "notes": [], "revision": 1,
+        }
+        _check(record)
+        stored = request.store.put(NOTEBOOKS, record)
+        remember(request.store, FAMILY, key, stored["id"])
+    request.audit("research.notebook_create", stored["id"])
+    return Response(201, stored, {"ETag": "1"})
+
+
+def get_notebook(request: Request) -> Response:
+    """``GET /hux/v1/notebooks/{id}``: one notebook."""
+    record = _resolve(request.store, NOTEBOOKS, request.params["id"])
+    request.audit("research.notebook_get", record["id"])
+    return Response(200, record, {"ETag": str(record["revision"])})
+
+
+def _merged_ids(store: TenantStore, family: str, current: list[str], body: dict[str, Any], key: str) -> list[str]:
+    added = _id_list(body, key, maximum=500)
+    for record_id in added:
+        _resolve(store, family, record_id)
+    return list(dict.fromkeys([*current, *added]))
+
+
+def patch_notebook(request: Request) -> Response:
+    """``PATCH /hux/v1/notebooks/{id}``: add references and notes, replace assumptions, move status (If-Match required)."""
+    body = _body(request)
+    expected = request.if_match()
+    if expected is None:
+        raise Invalid("If-Match is required")
+    with request.store.lock(FAMILY):
+        notebook = _resolve(request.store, NOTEBOOKS, request.params["id"])
+        if expected != notebook["revision"]:
+            raise Conflict(f"revision {expected} does not match current revision {notebook['revision']}", [str(notebook["revision"])])
+        stamp = now_iso()
+        updated = {
+            **notebook,
+            "source_ids": _merged_ids(request.store, SOURCES, notebook["source_ids"], body, "add_source_ids"),
+            "passage_ids": _merged_ids(request.store, PASSAGES, notebook["passage_ids"], body, "add_passage_ids"),
+            "citation_ids": _merged_ids(request.store, CITATIONS, notebook["citation_ids"], body, "add_citation_ids"),
+            "assumptions": _text_list(body, "assumptions", notebook["assumptions"]),
+            "unresolved_questions": _text_list(body, "unresolved_questions", notebook["unresolved_questions"]),
+            "updated_at": stamp,
+        }
+        notes = body.get("add_notes", [])
+        if not isinstance(notes, list) or not all(isinstance(n, dict) for n in notes):
+            raise Invalid("add_notes must be a list of objects")
+        for note in notes:
+            entry = {"at": stamp, "text": _string(note, "text", 2000)}
+            if note.get("source_id") is not None:
+                entry["source_id"] = _resolve(request.store, SOURCES, note["source_id"])["id"]
+            updated["notes"] = [*updated["notes"], entry]
+        status = body.get("status")
+        if status is not None and status != notebook["status"]:
+            if status not in NOTEBOOK_TRANSITIONS or status not in NOTEBOOK_TRANSITIONS[notebook["status"]]:
+                raise Conflict(f"notebook cannot move from {notebook['status']} to {status}")
+            updated["status"] = status
+        _check(updated)
+        stored = request.store.put(NOTEBOOKS, updated, expected_revision=expected)
+    request.audit("research.notebook_patch", stored["id"])
+    return Response(200, stored, {"ETag": str(stored["revision"])})
+
+
+def register(router: Router) -> None:
+    """Attach HUX-08 routes."""
+    card = "HUX-08"
+    router.add("POST", "/hux/v1/sources", card, "research.source_create", create_source)
+    router.add("GET", "/hux/v1/sources/{id}", card, "research.source_get", get_source)
+    router.add("POST", "/hux/v1/passages", card, "research.passage_create", create_passage)
+    router.add("POST", "/hux/v1/messages/{id}/citations", card, "research.citation_attach", attach_citation)
+    router.add("GET", "/hux/v1/messages/{id}/citations", card, "research.citation_list", list_citations)
+    router.add("POST", "/hux/v1/notebooks", card, "research.notebook_create", create_notebook)
+    router.add("GET", "/hux/v1/notebooks/{id}", card, "research.notebook_get", get_notebook)
+    router.add("PATCH", "/hux/v1/notebooks/{id}", card, "research.notebook_patch", patch_notebook)
diff --git a/testing/tests/test_hermes_hux_artifact_auth.py b/testing/tests/test_hermes_hux_artifact_auth.py
new file mode 100644
index 00000000..96c8e373
--- /dev/null
+++ b/testing/tests/test_hermes_hux_artifact_auth.py
@@ -0,0 +1,130 @@
+"""HUX-04 artifacts: authorization boundaries and tenant isolation.
+
+Security obligations exercised: a second subject sees 404, never 403, for
+every artifact route (SO-33, SO-34); malformed ids never reach a path (SO-43);
+a forged owner field inside the tenant directory is still refused; every
+denial leaves an audit outcome; the card serves nothing while its flag is off.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
+if str(FOUNDATION) not in sys.path:
+    sys.path.insert(0, str(FOUNDATION))
+
+from hux import artifacts, audit, contracts, errors, identity, store  # noqa: E402
+from hux.server import build_router  # noqa: E402
+
+SCHEMAS = contracts.load_all()
+OWNER = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat"}
+OTHER = {**OWNER, "X-Hux-Subject": "usr_fedcba9876543210"}
+ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
+
+
+@pytest.fixture
+def router(tmp_path):
+    return build_router(tmp_path, {"HUX_FLAGS": ALL_ON})
+
+
+def call(router, method, path, body=None, headers=OWNER):
+    payload = json.dumps(body).encode() if body is not None else b""
+    response = router.dispatch(method, path, headers, payload)
+    return response.status, response.body
+
+
+@pytest.fixture
+def artifact(router):
+    status, record = call(router, "POST", "/hux/v1/artifacts", {"type": "markdown", "title": "Mine", "content": "secret\n", "conversation_id": "conv_0001abcd"})
+    assert status == 201
+    return record
+
+
+def test_second_subject_gets_404_everywhere(router, artifact):
+    art = artifact["id"]
+    probes = [
+        ("GET", f"/hux/v1/artifacts/{art}", None),
+        ("GET", f"/hux/v1/artifacts/{art}/versions/1", None),
+        ("GET", f"/hux/v1/artifacts/{art}/versions/1/diff", None),
+        ("POST", f"/hux/v1/artifacts/{art}/versions", {"content": "x"}),
+        ("POST", f"/hux/v1/artifacts/{art}/promote", {"project_id": "prj_0001aaaa"}),
+    ]
+    for method, path, body in probes:
+        status, error = call(router, method, path, body, {**OTHER, "If-Match": "1"})
+        assert (status, error["code"]) == (404, "not_found"), path
+        assert contracts.validate_record(error, SCHEMAS) == []
+    status, body = call(router, "GET", "/hux/v1/artifacts?conversation_id=conv_0001abcd", headers=OTHER)
+    assert status == 200 and body["items"] == []
+    rows = audit.recent(store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_fedcba9876543210", "chat", "router")))
+    assert [r["outcome"] for r in rows if r["action"].startswith("artifacts.")].count("not_found") == 5
+
+
+def test_cross_tenant_lineage_forgery_is_404(router, artifact):
+    body = {"type": "markdown", "title": "Derived", "content": "copy", "lineage": {"artifact_id": artifact["id"], "version": 1}}
+    status, error = call(router, "POST", "/hux/v1/artifacts", body, OTHER)
+    assert (status, error["code"]) == (404, "not_found")
+    status, own = call(router, "POST", "/hux/v1/artifacts", {**body, "lineage": None}, OTHER)
+    assert status == 201
+    status, error = call(router, "POST", f"/hux/v1/artifacts/{own['id']}/versions", {"content": "v2", "lineage": {"artifact_id": artifact["id"], "version": 1}}, {**OTHER, "If-Match": "1"})
+    assert status == 404
+    assert call(router, "GET", f"/hux/v1/artifacts/{own['id']}", headers=OTHER)[1]["current_version"] == 1
+
+
+def test_malformed_ids_never_reach_the_filesystem(router):
+    for path in ("/hux/v1/artifacts/..", "/hux/v1/artifacts/art_x", "/hux/v1/artifacts/ART_0000000000"):
+        status, error = call(router, "GET", path)
+        assert status == 404 and error["code"] == "not_found", path
+    tenant = store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router"))
+    assert not (tenant.root / "artifacts" / "...json").exists()
+
+
+def test_forged_owner_inside_tenant_directory_is_refused(router, artifact):
+    tenant = store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router"))
+    forged = {**tenant.get("artifacts", artifact["id"]), "id": "art_forged000001", "owner": "usr_fedcba9876543210"}
+    tenant.put("artifacts", forged)
+    assert call(router, "GET", "/hux/v1/artifacts/art_forged000001")[0] == 404
+    status, error = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "t", "content": "c", "lineage": {"artifact_id": "art_forged000001", "version": 1}})
+    assert status == 404
+    status, body = call(router, "GET", "/hux/v1/artifacts")
+    assert [a["id"] for a in body["items"]] == [artifact["id"]]
+    assert artifacts.artifact_exists(tenant, "art_forged000001")
+
+
+def test_contract_validation_guards_every_write(router, artifact):
+    tenant = store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router"))
+    current = tenant.get("artifacts", artifact["id"])
+    with pytest.raises(errors.Invalid) as raised:
+        artifacts.append_version(tenant, current, {"version": 2, "created_at": "now", "created_by": {"type": "user", "id": "x"}, "content_ref": {"hash": "sha256:" + "a" * 64, "bytes": 1, "mime": "x"}}, 1)
+    assert raised.value.details and tenant.get("artifacts", artifact["id"])["current_version"] == 1
+
+
+def test_flag_off_hides_the_card(tmp_path, artifact):
+    router = build_router(tmp_path, {"HUX_FLAGS": "hux.foundation"})
+    status, error = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "t", "content": "c"})
+    assert (status, error["code"]) == (404, "flag_off")
+    status, error = call(router, "GET", "/hux/v1/artifacts")
+    assert status == 404
+
+
+def test_idempotency_replay_is_scoped_to_the_caller(router):
+    body = {"type": "code", "title": "t", "content": "c"}
+    status, mine = call(router, "POST", "/hux/v1/artifacts", body, {**OWNER, "Idempotency-Key": "shared-key-01"})
+    status, theirs = call(router, "POST", "/hux/v1/artifacts", body, {**OTHER, "Idempotency-Key": "shared-key-01"})
+    assert status == 201 and theirs["id"] != mine["id"] and theirs["owner"] == "usr_fedcba9876543210"
+    status, again = call(router, "POST", "/hux/v1/artifacts", body, {**OWNER, "Idempotency-Key": "shared-key-02"})
+    assert status == 201 and again["id"] != mine["id"]
+    status, error = call(router, "POST", f"/hux/v1/artifacts/{mine['id']}/versions", {"content": "v"}, {**OWNER, "If-Match": "1", "Idempotency-Key": "shared-key-02"})
+    assert status == 201
+
+
+def test_audit_rows_name_family_verbs(router, artifact):
+    call(router, "GET", f"/hux/v1/artifacts/{artifact['id']}")
+    rows = audit.recent(store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router")))
+    assert [r["action"] for r in rows] == ["artifacts.create", "artifacts.get"]
+    assert all(contracts.validate("common.schema.json", r, SCHEMAS, "/$defs/audit_outcome") == [] for r in rows)
diff --git a/testing/tests/test_hermes_hux_artifact_versions.py b/testing/tests/test_hermes_hux_artifact_versions.py
new file mode 100644
index 00000000..9801c06f
--- /dev/null
+++ b/testing/tests/test_hermes_hux_artifact_versions.py
@@ -0,0 +1,319 @@
+"""HUX-04 artifacts: immutable versions, diffs, lineage, promotion and caps.
+
+Security obligations exercised: server-side sha256 with mismatch rejection and
+never-rewritten blobs (SO-30), per-version / per-artifact / per-tenant caps
+(SO-31), nosniff attachment delivery (SO-32), lineage resolved under the
+caller's subtree (SO-33), owner-only access (SO-34) and If-Match on every
+revisioned write (SO-44).
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import sys
+import threading
+import types
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
+if str(FOUNDATION) not in sys.path:
+    sys.path.insert(0, str(FOUNDATION))
+
+from hux import artifacts, contracts, diffs, errors, identity, store  # noqa: E402
+from hux.server import build_router  # noqa: E402
+
+SCHEMAS = contracts.load_all()
+HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat"}
+ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
+
+
+def ident() -> identity.Identity:
+    return identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router")
+
+
+@pytest.fixture
+def router(tmp_path):
+    return build_router(tmp_path, {"HUX_FLAGS": ALL_ON})
+
+
+def call(router, method, path, body=None, headers=None, raw=None):
+    payload = raw if raw is not None else (json.dumps(body).encode() if body is not None else b"")
+    response = router.dispatch(method, path, {**HEADERS, **(headers or {})}, payload)
+    return response.status, response.body, response.headers
+
+
+def valid(record) -> bool:
+    return contracts.validate_record(record, SCHEMAS) == []
+
+
+def make(router, text="# one\nline\n", **extra):
+    body = {"type": "markdown", "title": "Doc", "content": text, "mime": "text/markdown", "conversation_id": "conv_0001abcd", **extra}
+    status, record, _ = call(router, "POST", "/hux/v1/artifacts", body)
+    assert status == 201, record
+    return record
+
+
+@pytest.fixture
+def events(monkeypatch):
+    """Fake events lane capturing emit() calls with the agreed signature."""
+    calls = []
+    module = types.ModuleType("hux.events")
+    module.emit = lambda store, identity, conversation_id, kind, summary, detail=None, evidence=None, sensitivity="personal", run_id=None, turn=None, correlation_id=None: calls.append((conversation_id, kind, summary, detail, evidence, sensitivity))
+    monkeypatch.setitem(sys.modules, "hux.events", module)
+    return calls
+
+
+# --- create ------------------------------------------------------------------------
+
+def test_create_computes_hash_and_serves_contract_valid_record(router, events):
+    record = make(router, language="markdown", message_id="msg-1", note="first")
+    assert valid(record)
+    digest = hashlib.sha256(b"# one\nline\n").hexdigest()
+    version = record["versions"][0]
+    assert version["content_ref"] == {"hash": f"sha256:{digest}", "bytes": 11, "mime": "text/markdown"}
+    assert version["created_by"] == {"type": "user", "id": "usr_0123456789abcdef"}
+    assert (version["message_id"], version["note"], record["language"]) == ("msg-1", "first", "markdown")
+    assert record["owner"] == "usr_0123456789abcdef" and record["access"] == {"mode": "owner"}
+    assert record["current_version"] == 1 and record["revision"] == 1
+    assert events[0][:2] == ("conv_0001abcd", "artifact.created") and events[0][5] == "personal"
+    assert events[0][4][0]["kind"] == "artifact_version"
+    assert store.TenantStore(router.data_root, ident()).get_blob(digest) == b"# one\nline\n"
+
+
+def test_create_accepts_base64_and_verifies_claimed_hash(router):
+    data = bytes(range(256))
+    body = {"type": "image", "title": "Pixels", "content_base64": base64.b64encode(data).decode(), "mime": "image/png", "sensitivity": "sensitive", "project_id": "prj_0001aaaa"}
+    status, record, _ = call(router, "POST", "/hux/v1/artifacts", {**body, "hash": "sha256:" + hashlib.sha256(data).hexdigest()})
+    assert status == 201 and valid(record) and record["versions"][0]["content_ref"]["bytes"] == 256
+    assert record["sensitivity"] == "sensitive" and record["project_id"] == "prj_0001aaaa"
+    status, error, _ = call(router, "POST", "/hux/v1/artifacts", {**body, "hash": "sha256:" + "0" * 64})
+    assert (status, error["code"]) == (400, "invalid") and valid(error)
+    status, error, _ = call(router, "POST", "/hux/v1/artifacts", {**body, "content_base64": "not*base64"})
+    assert status == 400
+    status, error, _ = call(router, "POST", "/hux/v1/artifacts", {**body, "content": "both"})
+    assert status == 400
+    status, record, _ = call(router, "POST", "/hux/v1/artifacts", {"type": "audio", "title": "Beep", "content_base64": base64.b64encode(b"\x00\x01").decode()})
+    assert status == 201 and record["versions"][0]["content_ref"]["mime"] == "application/octet-stream"
+
+
+def test_worker_trust_is_recorded_as_system_actor(tmp_path):
+    router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_WORKER_KEY": "wk"})
+    headers = {"X-Hux-Trust": "worker", "X-Hux-Surface": "worker", "X-Hux-Relay-Key": "wk"}
+    status, record, _ = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "gen.py", "content": "print(1)\n"}, headers)
+    assert status == 201 and record["versions"][0]["created_by"] == {"type": "system", "id": "worker"}
+
+
+@pytest.mark.parametrize("bad", [
+    {"type": "binary"}, {"type": None}, {"sensitivity": "secret"}, {"title": ""}, {"title": "x" * 201}, {"content": 5},
+    {"conversation_id": "../etc"}, {"lineage": "art_x"}, {"note": "n" * 201}, {"mime": ""},
+])
+def test_create_rejects_bad_bodies(router, bad):
+    body = {"type": "markdown", "title": "Doc", "content": "x", **bad}
+    status, error, _ = call(router, "POST", "/hux/v1/artifacts", body)
+    assert status == 400 and valid(error)
+    status, error, _ = call(router, "POST", "/hux/v1/artifacts", raw=b"[]")
+    assert status == 400
+
+
+def test_idempotency_key_replays_the_original(router):
+    body = {"type": "json", "title": "Data", "content": "{}"}
+    status, first, _ = call(router, "POST", "/hux/v1/artifacts", body, {"Idempotency-Key": "create-0001"})
+    status, again, headers = call(router, "POST", "/hux/v1/artifacts", body, {"Idempotency-Key": "create-0001"})
+    assert (status, headers.get("HUX-Replayed")) == (200, "true") and again["id"] == first["id"]
+    status, _, _ = call(router, "POST", "/hux/v1/artifacts", body, {"Idempotency-Key": "bad key"})
+    assert status == 400
+    assert store.TenantStore(router.data_root, ident()).count("artifacts") == 1
+
+
+# --- list / get ----------------------------------------------------------------------
+
+def test_list_filters_and_pages(router, monkeypatch):
+    monkeypatch.setattr(artifacts, "PAGE_SIZE", 2)
+    ids = [make(router)["id"] for _ in range(3)]
+    other = make(router, conversation_id="conv_0002abcd", project_id="prj_0001aaaa")["id"]
+    status, body, _ = call(router, "GET", "/hux/v1/artifacts?conversation_id=conv_0001abcd")
+    assert status == 200 and [a["id"] for a in body["items"]] == ids[:2] and body["next"] == "2"
+    status, body, _ = call(router, "GET", "/hux/v1/artifacts?conversation_id=conv_0001abcd&cursor=2")
+    assert [a["id"] for a in body["items"]] == ids[2:] and body["next"] is None
+    status, body, _ = call(router, "GET", "/hux/v1/artifacts?project_id=prj_0001aaaa")
+    assert [a["id"] for a in body["items"]] == [other] and all(valid(a) for a in body["items"])
+    assert call(router, "GET", "/hux/v1/artifacts?cursor=x")[0] == 400
+    status, record, headers = call(router, "GET", f"/hux/v1/artifacts/{other}")
+    assert status == 200 and record["id"] == other and headers["ETag"] == "1"
+
+
+# --- versions --------------------------------------------------------------------------
+
+def test_versions_append_immutably_with_if_match(router, events):
+    record = make(router)
+    path = f"/hux/v1/artifacts/{record['id']}/versions"
+    body = {"content": "# one\nline two\n", "mime": "text/markdown", "note": "edit", "message_id": "msg-2"}
+    assert call(router, "POST", path, body)[0] == 400
+    status, error, _ = call(router, "POST", path, body, {"If-Match": "7"})
+    assert (status, error["code"], error["details"]) == (409, "conflict", ["1"])
+    status, updated, headers = call(router, "POST", path, body, {"If-Match": "1"})
+    assert status == 201 and valid(updated) and headers["ETag"] == "2"
+    entry = updated["versions"][1]
+    assert (entry["version"], entry["diff_from"], entry["note"], entry["message_id"]) == (2, 1, "edit", "msg-2")
+    assert updated["current_version"] == 2 and updated["versions"][0] == record["versions"][0]
+    assert events[-1][1] == "artifact.version" and events[-1][3]["version"] == 2
+    status, third, _ = call(router, "POST", path, {"content": "third", "diff_from": 1}, {"If-Match": "2"})
+    assert third["versions"][2]["diff_from"] == 1
+    assert call(router, "POST", path, {"content": "x", "diff_from": 9}, {"If-Match": "3"})[0] == 404
+    assert call(router, "POST", path, {"content": "x", "diff_from": "1"}, {"If-Match": "3"})[0] == 400
+    status, error, _ = call(router, "POST", path, raw=b"[1]", headers={"If-Match": "3"})
+    assert status == 400
+
+
+def test_version_replay_and_old_blob_untouched(router):
+    record = make(router)
+    tenant = store.TenantStore(router.data_root, ident())
+    path = f"/hux/v1/artifacts/{record['id']}/versions"
+    status, second, _ = call(router, "POST", path, {"content": "v2"}, {"If-Match": "1", "Idempotency-Key": "ver-00001"})
+    status, replay, headers = call(router, "POST", path, {"content": "v2"}, {"If-Match": "1", "Idempotency-Key": "ver-00001"})
+    assert (status, headers["HUX-Replayed"], replay["revision"]) == (200, "true", 2)
+    old_digest = record["versions"][0]["content_ref"]["hash"].split(":")[1]
+    tenant.put_blob(old_digest, b"tampered")
+    assert tenant.get_blob(old_digest) == b"# one\nline\n"
+    with pytest.raises(errors.Conflict):
+        artifacts.append_version(tenant, second, dict(second["versions"][1]), 2)
+    with pytest.raises(errors.Conflict):
+        artifacts.append_version(tenant, second, {**second["versions"][1], "version": 5}, 2)
+    assert tenant.get("artifacts", record["id"])["versions"] == second["versions"]
+
+
+def test_concurrent_writers_never_lose_a_version(router):
+    record = make(router)
+    path = f"/hux/v1/artifacts/{record['id']}/versions"
+    statuses = []
+
+    def worker(n: int) -> None:
+        for _ in range(6):
+            _, current, _ = call(router, "GET", f"/hux/v1/artifacts/{record['id']}")
+            status, _, _ = call(router, "POST", path, {"content": f"w{n}"}, {"If-Match": str(current["revision"])})
+            statuses.append(status)
+
+    threads = [threading.Thread(target=worker, args=(n,)) for n in range(3)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    _, final, _ = call(router, "GET", f"/hux/v1/artifacts/{record['id']}")
+    assert statuses.count(201) == final["current_version"] - 1
+    assert [v["version"] for v in final["versions"]] == list(range(1, final["current_version"] + 1))
+    assert final["revision"] == final["current_version"]
+
+
+def test_lineage_resolves_only_under_the_caller(router):
+    parent = make(router)
+    child = make(router, lineage={"artifact_id": parent["id"], "version": 1})
+    assert child["versions"][0]["lineage"] == {"artifact_id": parent["id"], "version": 1}
+    path = f"/hux/v1/artifacts/{child['id']}/versions"
+    status, updated, _ = call(router, "POST", path, {"content": "v2", "lineage": {"artifact_id": parent["id"], "version": 1}}, {"If-Match": "1"})
+    assert status == 201 and valid(updated) and updated["versions"][1]["lineage"]["artifact_id"] == parent["id"]
+    for bad in ({"artifact_id": parent["id"], "version": 9}, {"artifact_id": "art_missing0000", "version": 1}, {"artifact_id": "../x", "version": 1}):
+        status, error, _ = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "t", "content": "c", "lineage": bad})
+        assert (status, error["code"]) == (404, "not_found"), bad
+    status, error, _ = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "t", "content": "c", "lineage": {"artifact_id": parent["id"], "version": "1"}})
+    assert status == 400
+
+
+def test_caps_are_413(router, monkeypatch):
+    record = make(router)
+    monkeypatch.setattr(artifacts, "MAX_VERSION_BYTES", 8)
+    status, error, _ = call(router, "POST", f"/hux/v1/artifacts/{record['id']}/versions", {"content": "123456789"}, {"If-Match": "1"})
+    assert (status, error["code"]) == (413, "too_large")
+    monkeypatch.setattr(artifacts, "MAX_VERSION_BYTES", 25 * 1024 * 1024)
+    monkeypatch.setattr(artifacts, "MAX_VERSIONS", 2)
+    assert call(router, "POST", f"/hux/v1/artifacts/{record['id']}/versions", {"content": "v2"}, {"If-Match": "1"})[0] == 201
+    assert call(router, "POST", f"/hux/v1/artifacts/{record['id']}/versions", {"content": "v3"}, {"If-Match": "2"})[0] == 413
+    monkeypatch.setattr(artifacts, "MAX_ARTIFACTS", 1)
+    status, error, _ = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "t", "content": "c"})
+    assert (status, error["code"]) == (413, "too_large")
+
+
+# --- content and diffs ------------------------------------------------------------------
+
+def test_get_version_serves_text_or_base64_with_nosniff(router):
+    record = make(router)
+    status, body, headers = call(router, "GET", f"/hux/v1/artifacts/{record['id']}/versions/1")
+    assert status == 200 and body["content"] == "# one\nline\n" and body["version"] == record["versions"][0]
+    assert headers["X-Content-Type-Options"] == "nosniff" and headers["Content-Disposition"] == "attachment"
+    status, body, _ = call(router, "POST", "/hux/v1/artifacts", {"type": "image", "title": "i", "content_base64": base64.b64encode(b"\xff\xfe").decode()})
+    status, body, _ = call(router, "GET", f"/hux/v1/artifacts/{body['id']}/versions/1")
+    assert base64.b64decode(body["content_base64"]) == b"\xff\xfe" and "content" not in body
+    status, body, _ = call(router, "POST", "/hux/v1/artifacts", {"type": "code", "title": "bin", "content_base64": base64.b64encode(b"\xff\xfe").decode()})
+    status, body, _ = call(router, "GET", f"/hux/v1/artifacts/{body['id']}/versions/1")
+    assert "content_base64" in body
+    assert call(router, "GET", f"/hux/v1/artifacts/{record['id']}/versions/2")[0] == 404
+    assert call(router, "GET", f"/hux/v1/artifacts/{record['id']}/versions/0")[0] == 400
+
+
+def test_diff_unified_for_text_and_hashes_for_binary(router):
+    record = make(router)
+    path = f"/hux/v1/artifacts/{record['id']}"
+    call(router, "POST", f"{path}/versions", {"content": "# one\nline two\n"}, {"If-Match": "1"})
+    call(router, "POST", f"{path}/versions", {"content": "# one\nline three\n"}, {"If-Match": "2"})
+    status, body, headers = call(router, "GET", f"{path}/versions/2/diff")
+    assert status == 200 and body["from"] == 1 and body["to"] == 2 and headers["X-Content-Type-Options"] == "nosniff"
+    assert "-line\n" in body["unified"] and "+line two\n" in body["unified"] and body["unified"].startswith("--- v1")
+    status, body, _ = call(router, "GET", f"{path}/versions/3/diff?from=1")
+    assert body["from"] == 1 and "+line three\n" in body["unified"]
+    status, body, _ = call(router, "GET", f"{path}/versions/1/diff")
+    assert body == {"from": 1, "to": 1, "unified": ""}
+    assert call(router, "GET", f"{path}/versions/3/diff?from=x")[0] == 400
+    assert call(router, "GET", f"{path}/versions/3/diff?from=9")[0] == 404
+    blob = base64.b64encode(b"\x00\x01").decode()
+    status, image, _ = call(router, "POST", "/hux/v1/artifacts", {"type": "image", "title": "i", "content_base64": blob})
+    call(router, "POST", f"/hux/v1/artifacts/{image['id']}/versions", {"content_base64": base64.b64encode(b"\x00\x01\x02").decode()}, {"If-Match": "1"})
+    status, body, _ = call(router, "GET", f"/hux/v1/artifacts/{image['id']}/versions/2/diff")
+    assert body["binary"]["from_bytes"] == 2 and body["binary"]["to_bytes"] == 3 and "unified" not in body
+    assert body["binary"]["to_hash"] == "sha256:" + hashlib.sha256(b"\x00\x01\x02").hexdigest()
+    assert diffs.unified("code", b"\xff", b"ok", 1, 2)["binary"]["from_bytes"] == 1
+
+
+# --- promotion ---------------------------------------------------------------------------
+
+def test_promote_sets_promotion_and_emits(router, events):
+    record = make(router)
+    call(router, "POST", f"/hux/v1/artifacts/{record['id']}/versions", {"content": "v2"}, {"If-Match": "1"})
+    path = f"/hux/v1/artifacts/{record['id']}/promote"
+    status, error, _ = call(router, "POST", path, {"project_id": "prj_0001aaaa"}, {"If-Match": "1"})
+    assert status == 409
+    status, promoted, headers = call(router, "POST", path, {"project_id": "prj_0001aaaa"}, {"If-Match": "2"})
+    assert status == 200 and valid(promoted) and headers["ETag"] == "3"
+    assert promoted["promotion"]["project_id"] == "prj_0001aaaa" and promoted["promotion"]["version"] == 2
+    assert promoted["project_id"] == "prj_0001aaaa"
+    assert events[-1][1] == "artifact.promoted" and events[-1][3]["project_id"] == "prj_0001aaaa"
+    status, promoted, _ = call(router, "POST", path, {"project_id": "prj_0002aaaa", "version": 1})
+    assert status == 200 and promoted["promotion"]["version"] == 1
+    assert call(router, "POST", path, {"project_id": "nope"})[0] == 400
+    assert call(router, "POST", path, {"project_id": "prj_0002aaaa", "version": 7})[0] == 404
+    assert call(router, "POST", path, raw=b"1")[0] == 400
+    from hux import audit
+    rows = [r for r in audit.recent(store.TenantStore(router.data_root, ident())) if r["action"] == "artifacts.promote" and r["outcome"] == "allow"]
+    assert [r.get("reason", "") for r in rows][-2:] == ["", "unconditional_write"]
+
+
+# --- helpers for other lanes and hygiene ---------------------------------------------------
+
+def test_helpers_and_event_fallbacks(router, monkeypatch):
+    record = make(router)
+    tenant = store.TenantStore(router.data_root, ident())
+    assert artifacts.artifact_exists(tenant, record["id"]) and not artifacts.artifact_exists(tenant, "bad id")
+    assert artifacts.artifact_titles(tenant, [record["id"], "art_missing0000"]) == {record["id"]: "Doc"}
+    monkeypatch.setitem(sys.modules, "hux.events", None)
+    artifacts.emit_event(tenant, ident(), record, "artifact.created", "s", {})
+    artifacts.emit_event(tenant, ident(), {**record, "conversation_id": None}, "artifact.created", "s", {})
+    status, plain, _ = call(router, "POST", "/hux/v1/artifacts", {"type": "csv", "title": "no conv", "content": "a,b\n"})
+    assert status == 201 and "conversation_id" not in plain
+
+
+def test_modules_stay_under_500_lines():
+    for name in ("artifacts.py", "diffs.py"):
+        assert len((FOUNDATION / "hux" / name).read_text().splitlines()) <= 500
diff --git a/testing/tests/test_hermes_hux_research_citations.py b/testing/tests/test_hermes_hux_research_citations.py
new file mode 100644
index 00000000..a507b640
--- /dev/null
+++ b/testing/tests/test_hermes_hux_research_citations.py
@@ -0,0 +1,238 @@
+"""HUX-08 research: sources, passages and citations.
+
+Security obligations exercised: the service stores but never dereferences a
+source ``uri`` and only allowlisted schemes are accepted (SO-19); passage and
+citation references resolve under the caller's subtree and foreign ids are
+404 (SO-33); hashes and dedupe keys are server-computed (SO-30 by analogy);
+every served record satisfies citation.schema.json.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
+if str(FOUNDATION) not in sys.path:
+    sys.path.insert(0, str(FOUNDATION))
+
+from hux import audit, contracts, identity, research, store  # noqa: E402
+from hux.server import build_router  # noqa: E402
+
+SCHEMAS = contracts.load_all()
+OWNER = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat"}
+OTHER = {**OWNER, "X-Hux-Subject": "usr_fedcba9876543210"}
+ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
+
+
+def tenant(router, subject="usr_0123456789abcdef") -> store.TenantStore:
+    return store.TenantStore(router.data_root, identity.Identity("slot-3", subject, "chat", "router"))
+
+
+@pytest.fixture
+def router(tmp_path):
+    return build_router(tmp_path, {"HUX_FLAGS": ALL_ON})
+
+
+def call(router, method, path, body=None, headers=None, raw=None):
+    payload = raw if raw is not None else (json.dumps(body).encode() if body is not None else b"")
+    response = router.dispatch(method, path, {**OWNER, **(headers or {})}, payload)
+    return response.status, response.body, response.headers
+
+
+def valid(record) -> bool:
+    return contracts.validate_record(record, SCHEMAS) == []
+
+
+def source(router, uri="https://Example.com/cabinets/#lead", **extra):
+    status, record, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": uri, "title": "Cabinet pricing", "classification": "primary", **extra})
+    assert status == 201, record
+    return record
+
+
+def passage(router, source_id, text="Standard base cabinets ship in 3-4 weeks."):
+    status, record, _ = call(router, "POST", "/hux/v1/passages", {"source_id": source_id, "text": text})
+    assert status == 201, record
+    return record
+
+
+def cite(router, passage_ids, message="msg-44", claim="Base cabinets take three to four weeks.", **extra):
+    status, record, _ = call(router, "POST", f"/hux/v1/messages/{message}/citations", {"claim": claim, "passage_ids": passage_ids, "support": "supports", **extra})
+    assert status == 201, record
+    return record
+
+
+@pytest.fixture
+def events(monkeypatch):
+    calls = []
+    module = types.ModuleType("hux.events")
+    module.emit = lambda store, identity, conversation_id, kind, summary, detail=None, evidence=None, sensitivity="personal", run_id=None, turn=None, correlation_id=None: calls.append((conversation_id, kind, detail, evidence))
+    monkeypatch.setitem(sys.modules, "hux.events", module)
+    return calls
+
+
+# --- sources -----------------------------------------------------------------------
+
+def test_source_is_valid_and_deduped_on_normalised_uri(router):
+    first = source(router, publisher="Example Co", published_at="2026-01-01T00:00:00Z", content_hash="sha256:" + "a" * 64, conversation_id="conv_0001abcd")
+    assert valid(first) and first["provenance"]["actor"] == {"type": "user", "id": "usr_0123456789abcdef"}
+    assert first["provenance"]["conversation_id"] == "conv_0001abcd" and first["publisher"] == "Example Co"
+    expected = "sha256:" + hashlib.sha256(b"uri\nhttps://example.com/cabinets").hexdigest()
+    assert first["dedupe_key"] == expected
+    status, again, headers = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "HTTPS://example.com/cabinets", "title": "Different title"})
+    assert (status, again["id"], headers["HUX-Replayed"]) == (200, first["id"], "true")
+    assert tenant(router).count("sources") == 1
+    status, record, _ = call(router, "GET", f"/hux/v1/sources/{first['id']}")
+    assert status == 200 and record == first
+    assert call(router, "GET", "/hux/v1/sources/src_missing00001")[0] == 404
+    assert call(router, "GET", "/hux/v1/sources/..")[0] == 404
+
+
+def test_source_without_uri_dedupes_on_kind_and_title(router):
+    status, memory, _ = call(router, "POST", "/hux/v1/sources", {"kind": "memory", "title": "  Remembered fact "})
+    status, again, _ = call(router, "POST", "/hux/v1/sources", {"kind": "memory", "title": "remembered fact"})
+    assert status == 200 and again["id"] == memory["id"] and "uri" not in memory and memory["classification"] == "unknown"
+    assert research.normalise_uri("artifact://art_0001aaaa@2/") == "artifact://art_0001aaaa@2/"
+    assert research.normalise_uri("file:///opt/data/x/") == "file:///opt/data/x"
+    assert research.normalise_uri("https://h.example/") == "https://h.example/"
+
+
+@pytest.mark.parametrize("bad", [
+    {"uri": "javascript:alert(1)"}, {"uri": "ftp://x"}, {"uri": "/relative"}, {"kind": "rumour"}, {"kind": None},
+    {"classification": "gospel"}, {"title": ""}, {"conversation_id": "x"}, {"content_hash": "md5:abc"}, {"published_at": "yesterday"},
+])
+def test_source_rejects_bad_bodies(router, bad):
+    status, error, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "https://ok.example/", "title": "t", **bad})
+    assert status == 400 and valid(error), bad
+    assert call(router, "POST", "/hux/v1/sources", raw=b'"str"')[0] == 400
+
+
+def test_source_idempotency_and_cap(router, monkeypatch):
+    status, one, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "https://a.example/1", "title": "t"}, {"Idempotency-Key": "src-key-0001"})
+    status, two, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "https://a.example/2", "title": "t"}, {"Idempotency-Key": "src-key-0001"})
+    assert status == 200 and two["id"] == one["id"]
+    monkeypatch.setattr(research, "MAX_SOURCES", 1)
+    status, error, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "https://a.example/3", "title": "t"})
+    assert (status, error["code"]) == (413, "too_large")
+
+
+def test_service_has_no_outbound_client():
+    import hux.artifacts
+    import hux.research
+    for name in ("urllib.request", "http.client", "socket"):
+        assert name not in sys.modules or not any(name in getattr(m, "__dict__", {}) for m in (hux.research, hux.artifacts))
+    text = (FOUNDATION / "hux" / "research.py").read_text()
+    assert "urlopen" not in text and "http.client" not in text
+
+
+# --- passages ----------------------------------------------------------------------
+
+def test_passage_hash_is_server_computed_and_deduped(router, monkeypatch):
+    src = source(router)
+    first = passage(router, src["id"])
+    assert valid(first) and first["hash"] == "sha256:" + hashlib.sha256(b"Standard base cabinets ship in 3-4 weeks.").hexdigest()
+    status, again, headers = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "Standard base cabinets ship in 3-4 weeks.", "hash": "sha256:" + "0" * 64})
+    assert (status, again["id"], headers["HUX-Replayed"]) == (200, first["id"], "true") and again["hash"] == first["hash"]
+    status, located, _ = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "Other", "locator": {"page": 2, "selector": "#x"}}, {"Idempotency-Key": "psg-key-00001"})
+    assert status == 201 and valid(located) and located["locator"] == {"page": 2, "selector": "#x"}
+    status, replay, _ = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "Ignored"}, {"Idempotency-Key": "psg-key-00001"})
+    assert status == 200 and replay["id"] == located["id"]
+    status, error, _ = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "Bad", "locator": {"page": 0}})
+    assert status == 400 and error["details"]
+    assert call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "x" * 4001})[0] == 400
+    assert call(router, "POST", "/hux/v1/passages", {"source_id": "src_missing00001", "text": "x"})[0] == 404
+    assert call(router, "POST", "/hux/v1/passages", {"text": "x"})[0] == 404
+    assert call(router, "POST", "/hux/v1/passages", raw=b"[]")[0] == 400
+    monkeypatch.setattr(research, "MAX_PASSAGES", 2)
+    assert call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "third"})[0] == 413
+
+
+def test_passage_cannot_name_another_subjects_source(router):
+    src = source(router)
+    status, error, _ = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "steal"}, OTHER)
+    assert (status, error["code"]) == (404, "not_found")
+    assert call(router, "GET", f"/hux/v1/sources/{src['id']}", headers=OTHER)[0] == 404
+    rows = audit.recent(tenant(router, "usr_fedcba9876543210"))
+    assert [r["outcome"] for r in rows] == ["not_found", "not_found"]
+
+
+# --- citations ---------------------------------------------------------------------
+
+def test_citation_attach_dedupes_and_lists_with_passages_and_sources(router, events):
+    src = source(router)
+    one = passage(router, src["id"])
+    two = passage(router, src["id"], "Delivery excludes installation.")
+    cit = cite(router, [two["id"], one["id"]], conversation_id="conv_0001abcd", note="checked")
+    assert valid(cit) and cit["passage_ids"] == [two["id"], one["id"]] and cit["note"] == "checked"
+    assert cit["dedupe_key"] == "sha256:" + hashlib.sha256(("\n".join(["citation", "msg-44", "Base cabinets take three to four weeks.", *sorted([one["id"], two["id"]])])).encode()).hexdigest()
+    assert events == [("conv_0001abcd", "citation.attached", {"message_id": "msg-44", "citation_id": cit["id"], "support": "supports"}, [{"kind": "passage", "id": two["id"], "hash": two["hash"]}, {"kind": "passage", "id": one["id"], "hash": one["hash"]}])]
+    status, again, headers = call(router, "POST", "/hux/v1/messages/msg-44/citations", {"claim": "Base cabinets take three to four weeks.", "passage_ids": [one["id"], two["id"], one["id"]], "support": "contradicts"})
+    assert (status, again["id"], headers["HUX-Replayed"]) == (200, cit["id"], "true") and again["support"] == "supports"
+    second = cite(router, [one["id"]], claim="Second claim")
+    status, body, _ = call(router, "GET", "/hux/v1/messages/msg-44/citations")
+    assert status == 200 and [i["citation"]["id"] for i in body["items"]] == [cit["id"], second["id"]]
+    strip = body["items"][0]
+    assert [p["id"] for p in strip["passages"]] == [two["id"], one["id"]] and strip["sources"] == [src]
+    assert all(valid(r) for item in body["items"] for r in (item["citation"], *item["passages"], *item["sources"]))
+    assert call(router, "GET", "/hux/v1/messages/msg-none/citations")[1]["items"] == []
+    assert len(events) == 1  # the second citation carried no conversation_id, so nothing to emit against
+
+
+@pytest.mark.parametrize("bad", [
+    {"claim": ""}, {"support": "maybe"}, {"passage_ids": "psg_x"}, {"passage_ids": []}, {"note": "n" * 501}, {"conversation_id": "nope"},
+])
+def test_citation_rejects_bad_bodies(router, bad):
+    src = source(router)
+    psg = passage(router, src["id"])
+    status, error, _ = call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "c", "passage_ids": [psg["id"]], "support": "supports", **bad})
+    assert status == 400 and valid(error), bad
+
+
+def test_citation_reference_rules(router):
+    src = source(router)
+    psg = passage(router, src["id"])
+    assert call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "c", "passage_ids": ["psg_missing00001"]})[0] == 404
+    assert call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "c", "passage_ids": [psg["id"]]}, OTHER)[0] == 404
+    status, error, _ = call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "c", "passage_ids": [f"psg_{i:012d}" for i in range(33)]})
+    assert status == 413
+    assert call(router, "POST", f"/hux/v1/messages/{'m' * 121}/citations", {"claim": "c", "passage_ids": [psg["id"]]})[0] == 400
+    assert call(router, "POST", "/hux/v1/messages/msg-1/citations", raw=b"null")[0] == 400
+    status, cit, _ = call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "c", "passage_ids": [psg["id"]]}, {"Idempotency-Key": "cit-key-00001"})
+    status, replay, _ = call(router, "POST", "/hux/v1/messages/msg-1/citations", {"claim": "other", "passage_ids": [psg["id"]]}, {"Idempotency-Key": "cit-key-00001"})
+    assert status == 200 and replay["id"] == cit["id"] and replay["support"] == "unverified"
+    assert call(router, "GET", "/hux/v1/messages/msg-1/citations", headers=OTHER)[1]["items"] == []
+
+
+def test_validate_citations_reports_integrity_problems(router, monkeypatch):
+    src = source(router)
+    psg = passage(router, src["id"])
+    good = cite(router, [psg["id"]])
+    contradiction = cite(router, [psg["id"]], claim="Never", support="contradicts")
+    noted = cite(router, [psg["id"]], claim="Never again", support="contradicts", note="supplier changed policy")
+    tenant_store = tenant(router)
+    assert research.validate_citations(tenant_store, [good["id"], noted["id"]]) == []
+    assert research.validate_citations(tenant_store, [contradiction["id"]]) == [f"{contradiction['id']}: contradicts without a note"]
+    tenant_store.delete("sources", src["id"])
+    assert research.validate_citations(tenant_store, [good["id"]]) == [f"{good['id']}: passage {psg['id']} names missing source {src['id']}"]
+    tenant_store.delete("passages", psg["id"])
+    problems = research.validate_citations(tenant_store, [good["id"], "cit_missing00001", "bad id"])
+    assert problems == [f"{good['id']}: passage {psg['id']} does not resolve", "cit_missing00001: citation does not resolve", "bad id: citation does not resolve"]
+    monkeypatch.setitem(sys.modules, "hux.events", None)
+    assert cite(router, [passage(router, source(router, "https://b.example/")["id"])["id"]], claim="silent", conversation_id="conv_0001abcd")["support"] == "supports"
+
+
+def test_research_modules_stay_under_500_lines():
+    assert len((FOUNDATION / "hux" / "research.py").read_text().splitlines()) <= 500
+
+
+def test_worker_trust_records_a_system_actor(tmp_path):
+    router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_WORKER_KEY": "wk"})
+    headers = {"X-Hux-Trust": "worker", "X-Hux-Surface": "worker", "X-Hux-Relay-Key": "wk"}
+    status, record, _ = call(router, "POST", "/hux/v1/sources", {"kind": "tool_output", "title": "grep"}, headers)
+    assert status == 201 and record["provenance"] == {"surface": "worker", "actor": {"type": "system", "id": "worker"}, "recorded_at": record["retrieved_at"]}
diff --git a/testing/tests/test_hermes_hux_research_notebook.py b/testing/tests/test_hermes_hux_research_notebook.py
new file mode 100644
index 00000000..0022a514
--- /dev/null
+++ b/testing/tests/test_hermes_hux_research_notebook.py
@@ -0,0 +1,162 @@
+"""HUX-08 research notebooks: state machine, references and concurrency.
+
+Security obligations exercised: every id added to a notebook resolves under
+the caller's subtree, so a second subject cannot attach or read another
+tenant's records (SO-33); PATCH requires If-Match and a stale revision is a
+409 (SO-44); status only moves open -> answered | abandoned; every served
+record satisfies citation.schema.json#/$defs/notebook.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
+if str(FOUNDATION) not in sys.path:
+    sys.path.insert(0, str(FOUNDATION))
+
+from hux import audit, contracts, identity, store  # noqa: E402
+from hux.server import build_router  # noqa: E402
+
+SCHEMAS = contracts.load_all()
+OWNER = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat"}
+OTHER = {**OWNER, "X-Hux-Subject": "usr_fedcba9876543210"}
+ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
+
+
+@pytest.fixture
+def router(tmp_path):
+    return build_router(tmp_path, {"HUX_FLAGS": ALL_ON})
+
+
+def call(router, method, path, body=None, headers=None, raw=None):
+    payload = raw if raw is not None else (json.dumps(body).encode() if body is not None else b"")
+    response = router.dispatch(method, path, {**OWNER, **(headers or {})}, payload)
+    return response.status, response.body, response.headers
+
+
+def valid(record) -> bool:
+    return contracts.validate_record(record, SCHEMAS) == []
+
+
+@pytest.fixture
+def evidence(router):
+    """One source, passage and citation owned by the default subject."""
+    _, src, _ = call(router, "POST", "/hux/v1/sources", {"kind": "web", "uri": "https://example.com/a", "title": "A"})
+    _, psg, _ = call(router, "POST", "/hux/v1/passages", {"source_id": src["id"], "text": "A says B."})
+    _, cit, _ = call(router, "POST", "/hux/v1/messages/msg-9/citations", {"claim": "B", "passage_ids": [psg["id"]], "support": "supports"})
+    return src, psg, cit
+
+
+def notebook(router, **extra):
+    status, record, headers = call(router, "POST", "/hux/v1/notebooks", {"conversation_id": "conv_0001abcd", "question": "Which supplier?", **extra})
+    assert status == 201, record
+    assert headers["ETag"] == "1"
+    return record
+
+
+def test_create_and_get_are_contract_valid(router):
+    record = notebook(router, assumptions=["Budget is fixed"], unresolved_questions=["Installation included?"])
+    assert valid(record) and record["status"] == "open" and record["revision"] == 1
+    assert record["assumptions"] == ["Budget is fixed"] and record["notes"] == []
+    status, fetched, headers = call(router, "GET", f"/hux/v1/notebooks/{record['id']}")
+    assert status == 200 and fetched == record and headers["ETag"] == "1"
+    status, replay, headers = call(router, "POST", "/hux/v1/notebooks", {"conversation_id": "conv_0001abcd", "question": "again"}, {"Idempotency-Key": "nb-key-000001"})
+    status, same, headers = call(router, "POST", "/hux/v1/notebooks", {"conversation_id": "conv_0001abcd", "question": "again"}, {"Idempotency-Key": "nb-key-000001"})
+    assert (status, same["id"], headers["HUX-Replayed"]) == (200, replay["id"], "true")
+
+
+@pytest.mark.parametrize("bad", [
+    {"conversation_id": "nope"}, {"conversation_id": None}, {"question": ""}, {"question": "q" * 1001},
+    {"assumptions": "one"}, {"assumptions": [""]}, {"unresolved_questions": [1]}, {"assumptions": ["a"] * 65},
+])
+def test_create_rejects_bad_bodies(router, bad):
+    status, error, _ = call(router, "POST", "/hux/v1/notebooks", {"conversation_id": "conv_0001abcd", "question": "q", **bad})
+    assert status == 400 and valid(error), bad
+    assert call(router, "POST", "/hux/v1/notebooks", raw=b"[]")[0] == 400
+
+
+def test_patch_adds_references_notes_and_lists(router, evidence):
+    src, psg, cit = evidence
+    record = notebook(router)
+    path = f"/hux/v1/notebooks/{record['id']}"
+    patch = {
+        "add_source_ids": [src["id"], src["id"]], "add_passage_ids": [psg["id"]], "add_citation_ids": [cit["id"]],
+        "add_notes": [{"text": "Lead times exclude installation.", "source_id": src["id"]}, {"text": "Plain note"}],
+        "assumptions": ["Budget cap 12k"], "unresolved_questions": [],
+    }
+    status, updated, headers = call(router, "PATCH", path, patch, {"If-Match": "1"})
+    assert status == 200 and valid(updated) and headers["ETag"] == "2"
+    assert updated["source_ids"] == [src["id"]] and updated["passage_ids"] == [psg["id"]] and updated["citation_ids"] == [cit["id"]]
+    assert [n["text"] for n in updated["notes"]] == ["Lead times exclude installation.", "Plain note"]
+    assert updated["notes"][0]["source_id"] == src["id"] and "source_id" not in updated["notes"][1]
+    assert updated["assumptions"] == ["Budget cap 12k"] and updated["unresolved_questions"] == []
+    status, again, _ = call(router, "PATCH", path, {"add_source_ids": [src["id"]]}, {"If-Match": "2"})
+    assert again["source_ids"] == [src["id"]] and again["revision"] == 3
+    status, error, _ = call(router, "PATCH", path, {"add_notes": [{"text": "x", "source_id": "src_missing00001"}]}, {"If-Match": "3"})
+    assert status == 404
+    for bad in ({"add_notes": "note"}, {"add_notes": [{"text": ""}]}, {"add_source_ids": "src_x"}, {"assumptions": [""]}):
+        status, error, _ = call(router, "PATCH", path, bad, {"If-Match": "3"})
+        assert status == 400 and valid(error), bad
+    assert call(router, "PATCH", path, raw=b"[]", headers={"If-Match": "3"})[0] == 400
+    status, error, _ = call(router, "PATCH", path, {"add_source_ids": [f"src_{i:012d}" for i in range(501)]}, {"If-Match": "3"})
+    assert status == 413
+
+
+def test_patch_requires_matching_if_match(router):
+    record = notebook(router)
+    path = f"/hux/v1/notebooks/{record['id']}"
+    status, error, _ = call(router, "PATCH", path, {"assumptions": ["x"]})
+    assert (status, error["code"]) == (400, "invalid")
+    status, error, _ = call(router, "PATCH", path, {"assumptions": ["x"]}, {"If-Match": "5"})
+    assert (status, error["code"], error["details"]) == (409, "conflict", ["1"])
+    assert call(router, "PATCH", path, {"assumptions": ["x"]}, {"If-Match": "abc"})[0] == 400
+    assert call(router, "GET", path)[1]["assumptions"] == []
+    rows = audit.recent(store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router")))
+    assert [r["outcome"] for r in rows if r["action"] == "research.notebook_patch"] == ["deny", "conflict", "deny"]
+
+
+@pytest.mark.parametrize(("target", "then", "expected"), [
+    ("answered", "abandoned", 409), ("abandoned", "answered", 409), ("answered", "open", 409), ("abandoned", "abandoned", 200),
+])
+def test_status_moves_only_from_open(router, target, then, expected):
+    record = notebook(router)
+    path = f"/hux/v1/notebooks/{record['id']}"
+    status, moved, _ = call(router, "PATCH", path, {"status": target}, {"If-Match": "1"})
+    assert status == 200 and moved["status"] == target and valid(moved)
+    status, body, _ = call(router, "PATCH", path, {"status": then}, {"If-Match": "2"})
+    assert status == expected
+    if expected == 409:
+        assert body["code"] == "conflict" and call(router, "GET", path)[1]["status"] == target
+    assert call(router, "PATCH", path, {"status": "done"}, {"If-Match": str(3 if expected == 200 else 2)})[0] == 409
+    assert call(router, "PATCH", path, {"status": target}, {"If-Match": str(3 if expected == 200 else 2)})[0] == 200
+
+
+def test_second_subject_cannot_read_patch_or_attach(router, evidence):
+    src, psg, cit = evidence
+    record = notebook(router)
+    path = f"/hux/v1/notebooks/{record['id']}"
+    assert call(router, "GET", path, headers=OTHER)[0] == 404
+    status, error, _ = call(router, "PATCH", path, {"status": "abandoned"}, {**OTHER, "If-Match": "1"})
+    assert (status, error["code"]) == (404, "not_found") and valid(error)
+    theirs = call(router, "POST", "/hux/v1/notebooks", {"conversation_id": "conv_0001abcd", "question": "spy"}, OTHER)[1]
+    for key, value in (("add_source_ids", src["id"]), ("add_passage_ids", psg["id"]), ("add_citation_ids", cit["id"])):
+        status, error, _ = call(router, "PATCH", f"/hux/v1/notebooks/{theirs['id']}", {key: [value]}, {**OTHER, "If-Match": "1"})
+        assert status == 404, key
+    status, mine, _ = call(router, "GET", path)
+    assert mine["status"] == "open" and mine["revision"] == 1
+    other_rows = audit.recent(store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_fedcba9876543210", "chat", "router")))
+    assert other_rows[-1]["outcome"] == "not_found" and other_rows[-1]["action"] == "research.notebook_patch"
+
+
+def test_notebook_paths_are_tenant_scoped(router):
+    record = notebook(router)
+    tenant = store.TenantStore(router.data_root, identity.Identity("slot-3", "usr_0123456789abcdef", "chat", "router"))
+    assert (tenant.root / "notebooks" / f"{record['id']}.json").exists()
+    assert call(router, "GET", "/hux/v1/notebooks/nb_missing000001")[0] == 404
+    assert call(router, "GET", "/hux/v1/notebooks/NB")[0] == 404