fix(register): require explicit topic when ambiguous

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a06d83-1cbc-71f2-b0dc-e0f48cedae43

custodian_cli.py

@@ -738,6 +738,11 @@ def main() -> None:
     )
     statehub_reg.add_argument("--path", default=os.getcwd(), help="Repo directory (defaults to cwd)")
     statehub_reg.add_argument("--domain", default=None, help="State Hub domain slug")
+    statehub_reg.add_argument(
+        "--topic",
+        default=None,
+        help="Existing active topic slug within the selected domain",
+    )
     statehub_reg.add_argument("--repo-slug", default=None, help="Repo slug (auto-detected if omitted)")
     statehub_reg.add_argument("--wp-prefix", default=None, help="Workplan prefix, e.g. STATE-WP")
     statehub_reg.add_argument("--description", default=None, help="One-sentence repo description")

statehub_register.py

@@ -250,7 +250,14 @@ def run_register(args: argparse.Namespace) -> None:
     if domain not in domain_slugs:
         domain = _ask_for_domain(domain, domain_slugs)
 
-    topic = _find_or_create_topic(domain, snapshot.project_name, repo_slug, inference, args.api_base)
+    topic = _find_or_create_topic(
+        domain,
+        snapshot.project_name,
+        repo_slug,
+        inference,
+        args.api_base,
+        requested_topic_slug=getattr(args, "topic", None),
+    )
     topic_id = topic["id"]
     topic_slug = topic.get("slug") or domain
 
@@ -831,11 +838,54 @@ def _find_or_create_topic(
     repo_slug: str,
     inference: RegisterInference,
     api_base: str,
+    *,
+    requested_topic_slug: str | None = None,
 ) -> dict[str, Any]:
     topics = _api_get("/topics/?status=active", api_base)
-    existing = next((t for t in topics if t.get("domain_slug") == domain), None)
+    domain_topics = [topic for topic in topics if topic.get("domain_slug") == domain]
-    if existing:
+
-        return existing
+    explicit_slug = _slugify(requested_topic_slug or "") or None
+    inferred_slug = inference.topic_slug
+    preferred_slug = explicit_slug or inferred_slug
+    if preferred_slug:
+        existing = next(
+            (topic for topic in domain_topics if topic.get("slug") == preferred_slug),
+            None,
+        )
+        if existing:
+            return existing
+
+    if explicit_slug:
+        cross_domain = [
+            str(topic.get("domain_slug"))
+            for topic in topics
+            if topic.get("slug") == explicit_slug and topic.get("domain_slug") != domain
+        ]
+        if cross_domain:
+            domains = ", ".join(sorted(set(cross_domain)))
+            raise SystemExit(
+                f"ERROR: Topic '{explicit_slug}' belongs to domain(s) {domains}, "
+                f"not '{domain}'."
+            )
+        available = ", ".join(
+            sorted(str(topic.get("slug")) for topic in domain_topics if topic.get("slug"))
+        ) or "(none)"
+        raise SystemExit(
+            f"ERROR: Active topic '{explicit_slug}' was not found in domain '{domain}'. "
+            f"Available: {available}"
+        )
+
+    if len(domain_topics) == 1:
+        return domain_topics[0]
+    if len(domain_topics) > 1:
+        available = ", ".join(
+            sorted(str(topic.get("slug")) for topic in domain_topics if topic.get("slug"))
+        )
+        raise SystemExit(
+            f"ERROR: Domain '{domain}' has multiple active topics: {available}. "
+            "Pass --topic <slug>."
+        )
+
     slug = inference.topic_slug or repo_slug
     title = inference.topic_title or project_name
     print(f"==> Creating active topic '{slug}' for domain '{domain}'")

tests/test_statehub_register_cli.py

@@ -16,6 +16,7 @@ from statehub_register import (
     project_registration_plan,
     refuse_project_flavor_scaffold,
     run_register,
+    _find_or_create_topic,
     _invoke_llm,
     _normalise_inference,
     _parse_json_object,
@@ -265,6 +266,7 @@ def _register_args(path: Path, **overrides):
         "repo_slug": None,
         "wp_prefix": None,
         "domain": "infotech",
+        "topic": None,
         "description": "Test project.",
         "intent": None,
         "api_base": "http://unused",
@@ -333,6 +335,120 @@ def test_run_register_existing_project_skips_scaffold_and_templating(
     assert not (repo / "INTENT.md").exists()
 
 
+def test_run_register_honors_explicit_topic_slug(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    repo = tmp_path / "demo-service"
+    repo.mkdir()
+    written, registered = _stub_register_io(monkeypatch)
+    monkeypatch.setattr(
+        "statehub_register._api_get",
+        lambda path, *_a, **_k: (
+            [{"slug": "infotech", "id": "dom-1"}]
+            if str(path).startswith("/domains")
+            else [
+                {"id": "topic-1", "slug": "custodian", "domain_slug": "infotech"},
+                {"id": "topic-2", "slug": "activity-core", "domain_slug": "infotech"},
+            ]
+        ),
+    )
+
+    run_register(
+        _register_args(
+            repo,
+            topic="activity-core",
+            intent="Register the demo service with the requested topic.",
+        )
+    )
+
+    assert written[0]["topic_id"] == "topic-2"
+    repo_registration = next(item for kind, item in registered if kind == "repo")
+    assert repo_registration["topic_id"] == "topic-2"
+
+
+def test_topic_selection_refuses_ambiguous_domain(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.setattr(
+        "statehub_register._api_get",
+        lambda *_a, **_k: [
+            {"id": "topic-1", "slug": "custodian", "domain_slug": "infotech"},
+            {"id": "topic-2", "slug": "activity-core", "domain_slug": "infotech"},
+        ],
+    )
+
+    with pytest.raises(SystemExit, match=r"Pass --topic <slug>"):
+        _find_or_create_topic(
+            "infotech",
+            "demo",
+            "demo",
+            RegisterInference(),
+            "http://unused",
+        )
+
+
+def test_topic_selection_uses_exact_inferred_topic(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    topics = [
+        {"id": "topic-1", "slug": "custodian", "domain_slug": "infotech"},
+        {"id": "topic-2", "slug": "activity-core", "domain_slug": "infotech"},
+    ]
+    monkeypatch.setattr("statehub_register._api_get", lambda *_a, **_k: topics)
+
+    selected = _find_or_create_topic(
+        "infotech",
+        "activity-core",
+        "activity-core",
+        RegisterInference(topic_slug="activity-core"),
+        "http://unused",
+    )
+
+    assert selected == topics[1]
+
+
+def test_topic_selection_refuses_cross_domain_topic(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.setattr(
+        "statehub_register._api_get",
+        lambda *_a, **_k: [
+            {"id": "topic-1", "slug": "activity-core", "domain_slug": "agents"},
+        ],
+    )
+
+    with pytest.raises(SystemExit, match=r"belongs to domain\(s\) agents"):
+        _find_or_create_topic(
+            "infotech",
+            "demo",
+            "demo",
+            RegisterInference(),
+            "http://unused",
+            requested_topic_slug="activity-core",
+        )
+
+
+def test_topic_selection_refuses_missing_explicit_topic(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.setattr(
+        "statehub_register._api_get",
+        lambda *_a, **_k: [
+            {"id": "topic-1", "slug": "custodian", "domain_slug": "infotech"},
+        ],
+    )
+
+    with pytest.raises(SystemExit, match=r"Available: custodian"):
+        _find_or_create_topic(
+            "infotech",
+            "demo",
+            "demo",
+            RegisterInference(),
+            "http://unused",
+            requested_topic_slug="activity-core",
+        )
+
+
 def test_write_registration_files_is_idempotent_without_force(tmp_path: Path):
     inference = RegisterInference()
     kwargs = {

workplans/ADHOC-2026-09-04.md

@@ -0,0 +1,34 @@
+---
+id: STATE-WP-ADHOC-2026-09-04
+type: workplan
+title: "Ad hoc fixes — 2026-09-04"
+domain: infotech
+repo: state-hub
+status: finished
+owner: codex
+topic_slug: state-hub
+created: "2026-09-04"
+updated: "2026-09-04"
+---
+
+## Add explicit topic selection to statehub register
+
+```task
+id: STATE-WP-ADHOC-2026-09-04-T01
+status: done
+priority: high
+```
+
+Add a `--topic` option that selects an existing active topic by slug within the
+chosen domain. Refuse missing, cross-domain, or ambiguous topic selection rather
+than silently assigning the first active topic in the domain. Preserve the
+single-topic fallback and exact inferred-topic selection for compatibility.
+
+Completed 2026-09-04. `statehub register --domain <domain> --topic <slug>` now
+selects the exact active topic and propagates its UUID to generated files, the
+repository registration/update, and the progress receipt. An explicit missing
+or cross-domain slug fails with a targeted error. Without `--topic`, an exact
+inference match wins, a sole domain topic remains backward-compatible, and an
+ambiguous domain now requires an explicit choice. Focused registration tests
+pass (`26 passed`); the full suite passes (`835 passed`, one pre-existing
+SQLAlchemy warning). Source distribution and wheel builds succeed.