feat(railiance): adopt state-hub edge relay beachhead for WP-0015

Deploy actcore-statehub-edge-relay, point STATE_HUB_URL at it, retire the
bespoke state-hub bridge, and accept edge-relay queued write receipts in
report and ops evidence sinks.

tests/test_ops_evidence_sinks.py

@@ -9,8 +9,9 @@ from activity_core.ops_evidence_sinks import persist_ops_inventory_evidence
 
 
 class DummyResponse:
-    def __init__(self, payload: Any) -> None:
+    def __init__(self, payload: Any, *, status_code: int = 200) -> None:
         self.payload = payload
+        self.status_code = status_code
 
     def raise_for_status(self) -> None:
         return None

tests/test_railiance_ops_inventory_wiring.py

@@ -33,6 +33,7 @@ def _by_kind_name(kind: str, name: str) -> dict[str, Any]:
 def test_runtime_config_has_ops_inventory_placeholders() -> None:
     config = _by_kind_name("ConfigMap", "actcore-runtime-config")
 
+    assert config["data"]["STATE_HUB_URL"] == "http://actcore-statehub-edge-relay:8000"
     assert config["data"]["LLM_CONNECT_URL"] == (
         "http://llm-connect.activity-core.svc.cluster.local:8080"
     )
@@ -112,11 +113,9 @@ def test_external_configmap_projects_enabled_daily_wsjf_definition(tmp_path) ->
     assert definition.trigger_config["timezone"] == "Europe/Berlin"
     assert instruction["id"] == "daily-triage-report"
     assert instruction["max_tokens"] == 1800
-    assert "most 7 recommendations" in instruction["prompt"]
-    assert "fewer well-formed" in instruction["prompt"]
-    assert instruction["output_schema"] == (
-        "/etc/activity-core/schemas/daily-triage-report.json"
-    )
+    assert "at most 7" in instruction["prompt"]
+    assert "fewer well-formed recommendations" in instruction["prompt"]
+    assert instruction["output_schema"] == "activity-core://schemas/daily-triage-report.json"
     assert instruction["report_sinks"][0]["type"] == "working-memory"
     assert instruction["report_sinks"][1]["event_type"] == "daily_triage"
 
@@ -136,12 +135,36 @@ def test_ops_inventory_configmap_contains_probeable_inventory() -> None:
     assert inventory["policy"]["non_secret_inventory"] is True
     assert services["gitea"]["endpoints"][0]["id"] == "gitea-oci-registry"
     assert services["state-hub"]["endpoints"][0]["url"] == (
-        "http://actcore-state-hub-bridge:8000/state/health"
+        "http://actcore-statehub-edge-relay:8000/edge/health"
     )
     assert services["inter-hub"]["endpoints"][0]["id"] == "inter-hub-openapi"
     assert services["activity-core"]["endpoints"][0]["id"] == "activity-core-api"
 
 
+def test_statehub_edge_relay_deployment_uses_state_hub_image_and_outbox_pvc() -> None:
+    deployment = _by_kind_name("Deployment", "actcore-statehub-edge-relay")
+    container = deployment["spec"]["template"]["spec"]["containers"][0]
+    env = {item["name"]: item["value"] for item in container["env"]}
+
+    assert container["image"] == "forgejo.coulomb.social/coulomb/state-hub:main-d8808bf"
+    assert env["STATEHUB_UPSTREAM_URL"] == "http://state-hub.state-hub.svc.cluster.local:8000"
+    assert env["STATEHUB_OUTBOX_PATH"] == "/var/statehub/edge-outbox.sqlite3"
+    assert env["STATEHUB_READ_CACHE_PATH"] == "/var/statehub/edge-read-cache.sqlite3"
+    assert deployment["spec"]["replicas"] == 1
+
+    pvc = _by_kind_name("PersistentVolumeClaim", "actcore-statehub-edge-outbox")
+    assert pvc["spec"]["resources"]["requests"]["storage"] == "1Gi"
+
+
+def test_legacy_state_hub_bridge_is_removed() -> None:
+    names = {
+        (resource.get("kind"), resource.get("metadata", {}).get("name"))
+        for resource in _resources()
+    }
+    assert ("Deployment", "actcore-state-hub-bridge") not in names
+    assert ("Service", "actcore-state-hub-bridge") not in names
+
+
 def test_worker_mounts_ops_inventory_configmap() -> None:
     deployment = _by_kind_name("Deployment", "actcore-worker")
     pod_spec = deployment["spec"]["template"]["spec"]
@@ -208,6 +231,8 @@ def test_disabled_ops_probe_definition_can_emit_fixture_evidence(
     def fake_endpoint_get(url: str, **kwargs: Any) -> Any:
         if url.endswith("/v2/"):
             return _HttpResponse(401, "OCI registry auth challenge")
+        if url.endswith("/edge/health"):
+            return _HttpResponse(200, "edge relay health")
         if url.endswith("/state/health"):
             return _HttpResponse(200, "health response")
         if url.endswith("/openapi.json"):
@@ -255,8 +280,9 @@ class _HttpResponse:
 
 
 class _JsonResponse:
-    def __init__(self, payload: Any) -> None:
+    def __init__(self, payload: Any, *, status_code: int = 200) -> None:
         self.payload = payload
+        self.status_code = status_code
 
     def raise_for_status(self) -> None:
         return None

tests/test_report_sinks.py

@@ -9,8 +9,9 @@ from activity_core.report_sinks import persist_reports
 
 
 class DummyResponse:
-    def __init__(self, payload: Any) -> None:
+    def __init__(self, payload: Any, *, status_code: int = 200) -> None:
         self.payload = payload
+        self.status_code = status_code
 
     def raise_for_status(self) -> None:
         return None

tests/test_state_hub_write.py

@@ -11,6 +11,7 @@ from activity_core.state_hub_write import (
     apply_progress_scope_fields,
     idempotency_headers,
     idempotency_key,
+    parse_state_hub_write_response,
 )
 
 
@@ -81,6 +82,8 @@ def test_report_sink_post_sends_idempotency_header(monkeypatch) -> None:
     monkeypatch.setattr(report_sinks, "_progress_exists", lambda *a, **k: False)
 
     class _Resp:
+        status_code = 200
+
         def raise_for_status(self) -> None: ...
         def json(self) -> dict[str, str]:
             return {"id": "pid-1"}
@@ -98,3 +101,47 @@ def test_report_sink_post_sends_idempotency_header(monkeypatch) -> None:
     result = report_sinks._post_state_hub_progress(payload, report_entry, sink)
     assert result["status"] == "posted"
     assert captured["headers"][IDEMPOTENCY_HEADER] == "run1:daily-triage-report:daily_triage"
+
+
+def test_parse_state_hub_write_response_accepts_edge_relay_queued_receipt() -> None:
+    class _Resp:
+        status_code = 202
+
+        def json(self) -> dict[str, object]:
+            return {
+                "queued": True,
+                "outbox_id": "env-1",
+                "idempotency_key": "run1:daily-triage-report:daily_triage",
+            }
+
+        def raise_for_status(self) -> None:
+            raise AssertionError("queued receipts should not raise")
+
+    assert parse_state_hub_write_response(_Resp())["outbox_id"] == "env-1"
+
+
+def test_report_sink_post_accepts_edge_relay_queued_receipt(monkeypatch) -> None:
+    monkeypatch.setattr(report_sinks, "_progress_exists", lambda *a, **k: False)
+
+    class _Resp:
+        status_code = 202
+
+        def json(self) -> dict[str, object]:
+            return {
+                "queued": True,
+                "outbox_id": "env-1",
+                "idempotency_key": "run1:daily-triage-report:daily_triage",
+            }
+
+        def raise_for_status(self) -> None:
+            raise AssertionError("queued receipts should not raise")
+
+    monkeypatch.setattr(report_sinks.httpx, "post", lambda *a, **k: _Resp())
+
+    payload = {"run_id": "run1", "activity_id": "act1", "scheduled_for": None}
+    report_entry = {"instruction_id": "daily-triage-report", "report": {"summary": "s"}}
+    sink = {"event_type": "daily_triage"}
+
+    result = report_sinks._post_state_hub_progress(payload, report_entry, sink)
+    assert result["status"] == "queued"
+    assert result["outbox_id"] == "env-1"