Link thread_uri to root node in POST /api/v1/threads

The API was creating orphan root nodes with no Uri record, so
the embed iframe could never find them. Now when thread_uri is
provided, a Uri record is created and linked to the node — same
association the embed uses via get_or_create_node_by_uri.

Returns 409 if a thread already exists for the given URI.

remarkbox/api/views.py

@@ -22,6 +22,7 @@ from remarkbox.models.namespace import (
     get_or_create_namespace,
     get_topsecret_namespaces,
 )
+from remarkbox.models.uri import get_or_create_uri
 from remarkbox.models.node import Node
 from remarkbox.lib.mail import send_verification_digits_to_email
 from remarkbox.lib.notify import schedule_notifications
@@ -443,6 +444,7 @@ def api_create_thread(request):
     namespace_name = body.get("namespace") or request.params.get("namespace")
     title = body.get("title") or request.params.get("thread_title", "")
     data = body.get("data") or request.params.get("thread_data", "")
+    thread_uri = body.get("thread_uri") or request.params.get("thread_uri", "")
     anonymous_name = (
         body.get("anonymous_name") or request.params.get("anonymous_name", "")
     ).strip()
@@ -528,6 +530,22 @@ def api_create_thread(request):
     request.dbsession.add(namespace)
     request.dbsession.flush()
 
+    # Link thread to a page URI so the embed iframe can find it
+    if thread_uri:
+        uri = get_or_create_uri(request.dbsession, thread_uri)
+        if uri.node is not None and uri.node.id != node.id:
+            # URI already has a thread — caller should use the reply endpoint
+            request.response.status_code = 409
+            return {
+                "error": "A thread already exists for this URI",
+                "existing_node_id": str(uri.node.id),
+            }
+        uri.node = node
+        node.has_uri = True
+        request.dbsession.add(uri)
+        request.dbsession.add(node)
+        request.dbsession.flush()
+
     if node_event:
         request.node = node
         schedule_notifications(request, node_event)

remarkbox/tests/test_api_views.py

@@ -97,6 +97,57 @@ class TestAPIAnonymousPosting(APIFunctionalTests):
         self.assertEqual(body["node"]["author"]["type"], "surrogate")
         self.assertEqual(body["node"]["author"]["name"], "ClaudeBot")
 
+    def test_create_thread_with_uri(self):
+        res = self.testapp.post_json(
+            "/api/v1/threads",
+            {
+                "namespace": self.namespace_name,
+                "title": "Page Thread",
+                "data": "Linked to a URI",
+                "thread_uri": "https://api-test.example.com/my-page/",
+                "anonymous_name": "TestBot",
+            },
+            expect_errors=True,
+        )
+        self.assertEqual(res.status_int, 201)
+        node_id = res.json["node"]["id"]
+
+        # Verify the URI record was created and linked
+        from remarkbox.models.uri import get_uri_by_uri
+        uri = get_uri_by_uri(self.dbsession, "https://api-test.example.com/my-page")
+        self.assertIsNotNone(uri)
+        self.assertEqual(str(uri.node.id), node_id)
+
+    def test_create_thread_duplicate_uri(self):
+        # First thread with URI
+        res1 = self.testapp.post_json(
+            "/api/v1/threads",
+            {
+                "namespace": self.namespace_name,
+                "title": "First",
+                "data": "First thread",
+                "thread_uri": "https://api-test.example.com/dup-test/",
+                "anonymous_name": "Bot",
+            },
+            expect_errors=True,
+        )
+        self.assertEqual(res1.status_int, 201)
+
+        # Second thread with same URI should fail
+        res2 = self.testapp.post_json(
+            "/api/v1/threads",
+            {
+                "namespace": self.namespace_name,
+                "title": "Second",
+                "data": "Duplicate",
+                "thread_uri": "https://api-test.example.com/dup-test/",
+                "anonymous_name": "Bot",
+            },
+            expect_errors=True,
+        )
+        self.assertEqual(res2.status_int, 409)
+        self.assertIn("already exists", res2.json["error"])
+
     def test_create_anonymous_thread_default_name(self):
         res = self.testapp.post_json(
             "/api/v1/threads",