Add hourly RecentlyOnScope batch endpoint

tests/test_recently_on_scope.py

@@ -171,3 +171,124 @@ class TestRecentlyOnScopeRoutes:
         response = await client.get("/domains/digest/recently-on-scope/20260522T120000Z--1h")
 
         assert response.status_code == 404
+
+    async def test_hourly_batch_generates_only_domains_with_activity(
+        self,
+        client,
+        tmp_path,
+        monkeypatch,
+        fake_markitect,
+    ):
+        monkeypatch.setattr(ros.settings, "state_hub_report_dir", str(tmp_path))
+
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        workstream = await _create_workstream(client, topic["id"])
+        task = await _create_task(client, workstream["id"], title="Batch source")
+        progress = await client.post(
+            "/progress/",
+            json={
+                "topic_id": topic["id"],
+                "workstream_id": workstream["id"],
+                "task_id": task["id"],
+                "event_type": "note",
+                "summary": "Batch source changed",
+            },
+        )
+        assert progress.status_code == 201, progress.text
+        await _create_domain(client, slug="quiet", name="Quiet Domain")
+
+        response = await client.post("/recently-on-scope/hourly", json={"range": "1d"})
+
+        assert response.status_code == 201, response.text
+        body = response.json()
+        assert [item["domain_slug"] for item in body["generated"]] == ["digest"]
+        assert [item["domain_slug"] for item in body["skipped"]] == ["quiet"]
+        assert body["failed"] == []
+        assert body["generated"][0]["source_counts"]["progress_events"] == 1
+        assert body["progress_event_id"] is not None
+
+        events = await client.get("/progress/", params={"event_type": "recently_on_scope_hourly"})
+        assert events.status_code == 200, events.text
+        assert events.json()[0]["detail"]["generated"][0]["domain_slug"] == "digest"
+
+    async def test_hourly_batch_reports_no_active_domains(
+        self,
+        client,
+        tmp_path,
+        monkeypatch,
+        fake_markitect,
+    ):
+        monkeypatch.setattr(ros.settings, "state_hub_report_dir", str(tmp_path))
+        await _create_domain(client, slug="quiet", name="Quiet Domain")
+
+        response = await client.post("/recently-on-scope/hourly", json={"range": "1d"})
+
+        assert response.status_code == 201, response.text
+        body = response.json()
+        assert body["generated"] == []
+        assert [item["domain_slug"] for item in body["skipped"]] == ["quiet"]
+        assert body["failed"] == []
+
+    async def test_hourly_batch_is_idempotent_for_exact_window(
+        self,
+        client,
+        tmp_path,
+        monkeypatch,
+        fake_markitect,
+    ):
+        monkeypatch.setattr(ros.settings, "state_hub_report_dir", str(tmp_path))
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        workstream = await _create_workstream(client, topic["id"])
+        await _create_task(client, workstream["id"], title="Idempotent source")
+        payload = {
+            "range": "1h",
+            "since": "2000-01-01T00:00:00Z",
+            "until": "2100-01-01T00:00:00Z",
+        }
+
+        first = await client.post("/recently-on-scope/hourly", json=payload)
+        second = await client.post("/recently-on-scope/hourly", json=payload)
+
+        assert first.status_code == 201, first.text
+        assert second.status_code == 201, second.text
+        assert first.json()["generated"][0]["id"] == "20000101T000000Z--21000101T000000Z"
+        listed = await client.get("/domains/digest/recently-on-scope/")
+        assert listed.status_code == 200
+        assert [item["id"] for item in listed.json()] == ["20000101T000000Z--21000101T000000Z"]
+
+    async def test_hourly_batch_continues_after_domain_failure(
+        self,
+        client,
+        tmp_path,
+        monkeypatch,
+        fake_markitect,
+    ):
+        monkeypatch.setattr(ros.settings, "state_hub_report_dir", str(tmp_path))
+        original_render = ros._render_report_from_data
+
+        def flaky_render(data, window):
+            if data["domain"]["slug"] == "broken":
+                raise RuntimeError("render failed")
+            return original_render(data, window)
+
+        monkeypatch.setattr(ros, "_render_report_from_data", flaky_render)
+
+        await _create_domain(client, slug="broken", name="Broken Domain")
+        broken_topic = await _create_topic(client, domain_slug="broken", slug="broken-topic")
+        broken_workstream = await _create_workstream(client, broken_topic["id"], slug="broken-ws")
+        await _create_task(client, broken_workstream["id"], title="Broken source")
+
+        await _create_domain(client, slug="good", name="Good Domain")
+        good_topic = await _create_topic(client, domain_slug="good", slug="good-topic")
+        good_workstream = await _create_workstream(client, good_topic["id"], slug="good-ws")
+        await _create_task(client, good_workstream["id"], title="Good source")
+
+        response = await client.post("/recently-on-scope/hourly", json={"range": "1d"})
+
+        assert response.status_code == 201, response.text
+        body = response.json()
+        assert [item["domain_slug"] for item in body["generated"]] == ["good"]
+        assert [item["domain_slug"] for item in body["failed"]] == ["broken"]
+        assert body["failed"][0]["error"] == "render failed"