HUB-WP-0006-T06: seed legacy topic aliases and add RM_SLICE_TOPICSPINE

- hub_core/runtime/data/legacy_topic_aliases.json: checked-in, versioned
  (2026-10-13.1) seed of the 16 State Hub topics from the STATE-WP-0098-T01
  capture -> primary_domain facet, residual owner STATE-WP-0079, expiry
  2027-03-31T23:59:59Z. Never fetched live; topic UUIDs kept verbatim.
- legacy_topic_aliases.py: loader validated against the 1.0.0
  legacyTopicAlias schema; per-response active/expired status; underscore
  slugs emitted hyphenated (1.0.0 pattern forbids '_').
- RM_SLICE_TOPICSPINE setting (off|shadow|on, default off). off keeps the
  navigation output unchanged (aliases []); shadow/on serve the aliases and
  mark responses with X-RM-Slice-Topicspine. Documented as the A5 rollback
  switch in docs/runtime.md.
- tests/test_legacy_topic_aliases.py: seed, flag and local A5 comparison
  cases (C01, C02/C08, C03, C10) against fixtures.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 356991@bnt-lap001
Assistant-Session: 5d99ea69-ed55-4a1e-8d97-f4f312db247f

tests/fixtures/a5-legacy-topics-2026-10-13.json

@@ -0,0 +1,103 @@
+{
+  "captured_at": "2026-10-13",
+  "request": "GET /topics/?status=active",
+  "source": "State Hub primary (read-only GET; mirrors STATE-WP-0098-T01 capture)",
+  "topics": [
+    {
+      "id": "084430ab-c630-48dc-9e1d-d07d1e8fce3c",
+      "slug": "personhood",
+      "domain_slug": "government",
+      "status": "active"
+    },
+    {
+      "id": "1f2e4d10-c967-4803-ae6c-7f4b4e806409",
+      "slug": "inter_hub",
+      "domain_slug": "infotech",
+      "status": "active"
+    },
+    {
+      "id": "36c7421b-c537-4723-bf75-42a3ebc6a1dc",
+      "slug": "coulomb-social",
+      "domain_slug": "communication",
+      "status": "active"
+    },
+    {
+      "id": "4c2e5315-2cb9-447c-9d16-a39bdb0aabd0",
+      "slug": "whynot",
+      "domain_slug": "consumer",
+      "status": "active"
+    },
+    {
+      "id": "5571d954-0d30-4950-980d-7bcaaad8e3e2",
+      "slug": "markitect",
+      "domain_slug": "communication",
+      "status": "active"
+    },
+    {
+      "id": "595afc64-bd28-47bf-aafb-ba230b28371b",
+      "slug": "stack",
+      "domain_slug": "infotech",
+      "status": "active"
+    },
+    {
+      "id": "64418556-3206-457a-ba29-6884b5b12cf3",
+      "slug": "foerster-capabilities",
+      "domain_slug": "agents",
+      "status": "active"
+    },
+    {
+      "id": "7d4d3e35-312f-4723-bde9-ddb43799109e",
+      "slug": "vergabe_teilnahme",
+      "domain_slug": "government",
+      "status": "active"
+    },
+    {
+      "id": "96fa8e80-9f74-40f2-84cd-644e9747b9ec",
+      "slug": "citation_evidence_mvp",
+      "domain_slug": "infotech",
+      "status": "active"
+    },
+    {
+      "id": "a6c6e745-bf54-4465-9340-1534a2be493e",
+      "slug": "netkingdom",
+      "domain_slug": "communication",
+      "status": "active"
+    },
+    {
+      "id": "c1d199b6-55ee-4db6-b49e-257a9f0f15ac",
+      "slug": "practice",
+      "domain_slug": "agents",
+      "status": "active"
+    },
+    {
+      "id": "ca369340-a64e-442e-98f1-a4fa7dc74a38",
+      "slug": "railiance",
+      "domain_slug": "financials",
+      "status": "active"
+    },
+    {
+      "id": "cee7bedf-2b48-46ef-8601-006474f2ad7a",
+      "slug": "custodian",
+      "domain_slug": "infotech",
+      "status": "active"
+    },
+    {
+      "id": "e61b8d01-78ec-47af-a68b-b92afa6fb188",
+      "slug": "nudge-space",
+      "domain_slug": "infotech",
+      "status": "active"
+    },
+    {
+      "id": "f39fa2a3-c491-414c-a91b-b4c5fcc6139c",
+      "slug": "helix-forge",
+      "domain_slug": "infotech",
+      "status": "active"
+    },
+    {
+      "id": "fe2aaa78-9c20-4feb-b3d2-4fe0529572a3",
+      "slug": "canon",
+      "domain_slug": "infotech",
+      "status": "active"
+    }
+  ]
+}

tests/test_legacy_topic_aliases.py

@@ -0,0 +1,267 @@
+"""Legacy topic alias seed and the RM_SLICE_TOPICSPINE A5 rollback switch.
+
+The ``test_a5_*`` cases run the State Hub A5 comparison packet
+(STATE-WP-0098-T01, cases A5-C01..C10) locally against checked-in fixtures:
+the legacy side is the 2026-10-13 ``GET /topics/?status=active`` capture and
+the candidate side is the in-memory runtime fed with the contract fixture page.
+"""
+
+from __future__ import annotations
+
+import json
+from copy import deepcopy
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from jsonschema import Draft202012Validator, FormatChecker
+
+from hub_core.contracts import repository_navigation_contract_root
+from hub_core.runtime.app import create_app
+from hub_core.runtime.config import RuntimeSettings
+from hub_core.runtime.legacy_topic_aliases import (
+    LegacyTopicAliasError,
+    load_legacy_topic_aliases,
+    parse_legacy_topic_aliases,
+)
+from hub_core.runtime.repository_navigation import RepositoryNavigationService
+from hub_core.runtime.store import InMemoryPortStore
+
+LEGACY_CAPTURE = Path(__file__).parent / "fixtures" / "a5-legacy-topics-2026-10-13.json"
+SEED = Path(__file__).parents[1] / "hub_core" / "runtime" / "data" / "legacy_topic_aliases.json"
+REPOSITORIES = "/ports/projections/repository-navigation/repositories"
+
+
+def legacy_topics() -> list[dict[str, Any]]:
+    return json.loads(LEGACY_CAPTURE.read_text(encoding="utf-8"))["topics"]
+
+
+def seed_document() -> dict[str, Any]:
+    return json.loads(SEED.read_text(encoding="utf-8"))
+
+
+def fixture_page() -> dict[str, Any]:
+    return json.loads(
+        repository_navigation_contract_root()
+        .joinpath("fixtures", "repository-classification-page.json")
+        .read_text(encoding="utf-8")
+    )
+
+
+def output_validator() -> Draft202012Validator:
+    schema = json.loads(
+        repository_navigation_contract_root()
+        .joinpath("schemas", "repository-navigation-projection.schema.json")
+        .read_text(encoding="utf-8")
+    )
+    return Draft202012Validator(schema, format_checker=FormatChecker())
+
+
+class PageClient:
+    def __init__(self, page: dict[str, Any]) -> None:
+        self.page = page
+
+    async def fetch_classification_page(self, cursor: str | None) -> dict[str, Any]:
+        return deepcopy(self.page)
+
+
+def runtime(mode: str | None) -> TestClient:
+    kwargs = {} if mode is None else {"rm_slice_topicspine": mode}
+    settings = RuntimeSettings(
+        environment="test", backend="memory", allow_ephemeral=True, **kwargs
+    )
+    return TestClient(
+        create_app(
+            settings=settings,
+            port_store=InMemoryPortStore(),
+            repo_projection_client=PageClient(fixture_page()),
+        )
+    )
+
+
+def as_legacy_rows(aliases: list[dict[str, Any]]) -> set[tuple[str, str, str]]:
+    return {
+        (alias["legacy_topic_id"], alias["legacy_topic_slug"], alias["facet_value"])
+        for alias in aliases
+    }
+
+
+def expected_rows() -> set[tuple[str, str, str]]:
+    # Normalisation documented in the seed: 1.0.0 slugs forbid underscores.
+    return {
+        (topic["id"], topic["slug"].replace("_", "-"), topic["domain_slug"])
+        for topic in legacy_topics()
+    }
+
+
+# --- seed ------------------------------------------------------------------
+
+
+def test_seed_is_versioned_non_authoritative_and_expiring() -> None:
+    document = seed_document()
+    alias_set = load_legacy_topic_aliases()
+
+    assert document["authority"] is False
+    assert alias_set.alias_set_version == document["alias_set_version"]
+    assert len(alias_set.aliases) == 16
+    assert all(alias.facet_kind == "primary_domain" for alias in alias_set.aliases)
+    assert all(alias.residual_owner == "STATE-WP-0079" for alias in alias_set.aliases)
+    assert all(
+        alias.expires_at == datetime(2027, 3, 31, 23, 59, 59, tzinfo=timezone.utc)
+        for alias in alias_set.aliases
+    )
+
+
+def test_seed_keeps_legacy_topic_ids_and_spelling_stable() -> None:
+    alias_set = load_legacy_topic_aliases()
+    by_id = {alias.legacy_topic_id: alias for alias in alias_set.aliases}
+
+    for topic in legacy_topics():
+        alias = by_id[topic["id"]]
+        assert alias.legacy_topic_slug == topic["slug"]
+        assert alias.facet_value == topic["domain_slug"]
+    assert alias_set.resolve("inter_hub") is alias_set.resolve("inter-hub")
+    assert alias_set.resolve("cee7bedf-2b48-46ef-8601-006474f2ad7a").facet_value == "infotech"
+    assert alias_set.resolve("no-such-topic") is None
+
+
+def test_alias_status_turns_expired_at_expires_at() -> None:
+    alias_set = load_legacy_topic_aliases()
+    before = alias_set.to_contract(now=datetime(2027, 3, 31, 23, 59, 58, tzinfo=timezone.utc))
+    after = alias_set.to_contract(now=datetime(2027, 3, 31, 23, 59, 59, tzinfo=timezone.utc))
+
+    assert {alias["status"] for alias in before} == {"active"}
+    assert {alias["status"] for alias in after} == {"expired"}
+
+
+@pytest.mark.parametrize(
+    ("mutate", "message"),
+    [
+        (lambda doc: doc.update(authority=True), "authority"),
+        (lambda doc: doc.pop("expires_at"), "expires_at"),
+        (lambda doc: doc.update(residual_owner=""), "legacyTopicAlias"),
+        (lambda doc: doc["aliases"].append(dict(doc["aliases"][0])), "duplicate"),
+        (lambda doc: doc["aliases"][0].update(facet_value="no-such-domain"), "invalid facet"),
+        (lambda doc: doc["aliases"][0].update(legacy_topic_id="not-a-uuid"), "legacyTopicAlias"),
+        (lambda doc: doc.update(aliases=[]), "non-empty"),
+    ],
+)
+def test_invalid_seed_is_rejected(mutate, message) -> None:
+    document = seed_document()
+    mutate(document)
+    with pytest.raises(LegacyTopicAliasError, match=message):
+        parse_legacy_topic_aliases(document)
+
+
+# --- RM_SLICE_TOPICSPINE ---------------------------------------------------
+
+
+def test_flag_defaults_off_and_reads_the_documented_env_name(monkeypatch) -> None:
+    monkeypatch.delenv("RM_SLICE_TOPICSPINE", raising=False)
+    assert RuntimeSettings().rm_slice_topicspine == "off"
+    assert RuntimeSettings.from_env().rm_slice_topicspine == "off"
+    monkeypatch.setenv("RM_SLICE_TOPICSPINE", " Shadow ")
+    assert RuntimeSettings.from_env().rm_slice_topicspine == "shadow"
+    monkeypatch.setenv("RM_SLICE_TOPICSPINE", "true")
+    with pytest.raises(ValueError, match="RM_SLICE_TOPICSPINE"):
+        RuntimeSettings.from_env()
+    with pytest.raises(ValueError, match="RM_SLICE_TOPICSPINE"):
+        RuntimeSettings(rm_slice_topicspine="maybe")
+
+
+def test_service_never_serves_aliases_while_off() -> None:
+    service = RepositoryNavigationService(
+        client=None,
+        store=InMemoryPortStore(),
+        legacy_topic_aliases=load_legacy_topic_aliases(),
+    )
+    assert service.topic_spine_mode == "off"
+    assert service.legacy_topic_aliases is None
+    with pytest.raises(ValueError, match="alias set"):
+        RepositoryNavigationService(client=None, store=InMemoryPortStore(), topic_spine_mode="on")
+
+
+def test_flag_off_leaves_navigation_output_unchanged() -> None:
+    with runtime(None) as default, runtime("off") as off:
+        responses = [client.get(REPOSITORIES) for client in (default, off)]
+        ready = off.get("/readyz").json()["checks"]
+
+    for response in responses:
+        assert response.status_code == 200
+        assert response.json()["legacy_topic_aliases"] == []
+        assert "x-rm-slice-topicspine" not in response.headers
+        assert "x-legacy-topic-alias-set" not in response.headers
+    strip = lambda body: {k: v for k, v in body.items() if k not in {"rebuilt_at", "source_checked_at"}}
+    assert strip(responses[0].json()) == strip(responses[1].json())
+    assert "rm_slice_topicspine" not in ready
+
+
+@pytest.mark.parametrize("mode", ["shadow", "on"])
+def test_flag_shadow_or_on_serves_contract_valid_aliases(mode: str) -> None:
+    with runtime(mode) as client:
+        listing = client.get(REPOSITORIES)
+        facet = client.get(
+            "/ports/projections/repository-navigation/facets/primary_domain/infotech"
+        )
+        ready = client.get("/readyz").json()["checks"]
+
+    for response in (listing, facet):
+        assert response.status_code == 200
+        output_validator().validate(response.json())
+        assert response.headers["x-rm-slice-topicspine"] == mode
+        assert response.headers["x-legacy-topic-alias-set"] == "2026-10-13.1"
+        assert len(response.json()["legacy_topic_aliases"]) == 16
+    # Classification rows are untouched by the alias metadata.
+    assert [row["slug"] for row in listing.json()["repositories"]] == ["hub-core", "repo-manager"]
+    assert ready["rm_slice_topicspine"] == mode
+
+
+# --- A5 comparison cases run locally against fixtures -----------------------
+
+
+def test_a5_c01_topic_list_maps_to_aliases() -> None:
+    with runtime("shadow") as client:
+        aliases = client.get(REPOSITORIES, params={"limit": 500}).json()["legacy_topic_aliases"]
+
+    assert as_legacy_rows(aliases) == expected_rows()
+    assert {alias["facet_kind"] for alias in aliases} == {"primary_domain"}
+    assert all(alias["expires_at"] and alias["residual_owner"] for alias in aliases)
+
+
+def test_a5_c02_and_c08_each_topic_id_resolves_to_one_alias() -> None:
+    with runtime("shadow") as client:
+        aliases = client.get(REPOSITORIES, params={"limit": 500}).json()["legacy_topic_aliases"]
+
+    for topic in legacy_topics():
+        matches = [alias for alias in aliases if alias["legacy_topic_id"] == topic["id"]]
+        assert len(matches) == 1
+        assert matches[0]["facet_value"] == topic["domain_slug"]
+    custodian = next(a for a in aliases if a["legacy_topic_id"] == "cee7bedf-2b48-46ef-8601-006474f2ad7a")
+    assert (custodian["legacy_topic_slug"], custodian["facet_value"]) == ("custodian", "infotech")
+
+
+def test_a5_c03_unknown_facet_value_is_rejected_not_empty() -> None:
+    # Packet expected 200 + zero repositories (unverified); the runtime
+    # validates the controlled domain vocabulary and answers 400 instead.
+    with runtime("shadow") as client:
+        unknown = client.get(
+            "/ports/projections/repository-navigation/facets/primary_domain/no-such-domain"
+        )
+        empty = client.get(
+            "/ports/projections/repository-navigation/facets/primary_domain/space"
+        )
+
+    assert unknown.status_code == 400
+    assert empty.status_code == 200
+    assert empty.json()["repositories"] == []
+
+
+def test_a5_c10_rollback_to_off_withdraws_aliases() -> None:
+    with runtime("on") as switched:
+        assert len(switched.get(REPOSITORIES).json()["legacy_topic_aliases"]) == 16
+    with runtime("off") as rolled_back:
+        response = rolled_back.get(REPOSITORIES)
+    assert response.json()["legacy_topic_aliases"] == []
+    assert "x-rm-slice-topicspine" not in response.headers