feat(cloud): enforce signed workspace subscription resource limits

tests/integration/api/test_workspaces.py

@@ -824,6 +824,7 @@ async def test_cloud_projection_is_selected_explicitly_and_collaboration_runs_in
                 not_before=1,
                 expires_at=4102444800,
                 plan_name='free',
+                limits={'bots.max': 2, 'pipelines.max': 3, 'plugins.max': 4, 'knowledge_bases.max': 5},
             )
 
     application.entitlement_resolver = PlanResolver()
@@ -843,6 +844,7 @@ async def test_cloud_projection_is_selected_explicitly_and_collaboration_runs_in
     assert current['workspace']['source'] == 'cloud_projection'
     assert current['placement_generation'] == 12
     assert current['plan_name'] == 'free'
+    assert current['limitation']['max_bots'] == 2
 
     create_workspace = await client.post(
         '/api/v1/workspaces',

tests/unit_tests/test_workspace_subscription_limits.py

@@ -0,0 +1,41 @@
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from langbot.pkg.cloud.entitlements import EntitlementSnapshot, EntitlementUnavailableError
+from langbot.pkg.api.http.service.workspace_limits import workspace_limits
+
+
+def test_oss_retains_instance_limits():
+    ap = SimpleNamespace(instance_config=SimpleNamespace(data={'system': {'limitation': {'max_bots': 7}}}))
+    assert asyncio.run(workspace_limits(ap, 'one')) == {'max_bots': 7}
+
+
+def test_cloud_uses_signed_workspace_limits():
+    class Resolver:
+        async def resolve(self, workspace_uuid, **kwargs):
+            assert workspace_uuid == 'one'
+            return EntitlementSnapshot(
+                instance_uuid='i',
+                workspace_uuid='one',
+                entitlement_revision=1,
+                status='active',
+                not_before=0,
+                expires_at=9999999999,
+                limits={'bots.max': 2, 'pipelines.max': 3, 'plugins.max': 4, 'knowledge_bases.max': 5},
+            )
+
+    ap = SimpleNamespace(deployment=SimpleNamespace(mode='cloud'), entitlement_resolver=Resolver())
+    assert asyncio.run(workspace_limits(ap, 'one')) == {
+        'max_bots': 2,
+        'max_pipelines': 3,
+        'max_extensions': 4,
+        'max_knowledge_bases': 5,
+    }
+
+
+def test_cloud_missing_resolver_fails_closed():
+    ap = SimpleNamespace(deployment=SimpleNamespace(mode='cloud'))
+    with pytest.raises(EntitlementUnavailableError):
+        asyncio.run(workspace_limits(ap, 'one'))