Use session workspace slug in product smoke

scripts/authenticated-product-smoke.py

@@ -178,6 +178,35 @@ def login(client: Client, email: str, password: str) -> None:
     )
 
 
+def resolve_workspace(
+    client: Client,
+    fallback_workspace_id: str,
+    fallback_workspace_slug: str,
+) -> tuple[str, str]:
+    session = client.request_json("GET", "/api/auth/session").body
+    memberships = session.get("memberships") or []
+    current_workspace_id = session.get("current_workspace_id") or fallback_workspace_id
+    membership = next(
+        (
+            item
+            for item in memberships
+            if item.get("workspace", {}).get("id") == current_workspace_id
+        ),
+        None,
+    )
+    if not membership and memberships:
+        membership = memberships[0]
+
+    workspace = membership.get("workspace", {}) if membership else {}
+    workspace_id = workspace.get("id") or current_workspace_id
+    workspace_slug = workspace.get("slug") or fallback_workspace_slug
+    if not workspace_id:
+        raise SmokeError("authenticated session does not include a workspace id")
+    if not workspace_slug:
+        raise SmokeError("authenticated session does not include a workspace slug")
+    return workspace_id, workspace_slug
+
+
 def create_operation(
     client: Client,
     workspace_id: str,
@@ -378,23 +407,29 @@ def run(args: argparse.Namespace) -> None:
     print(f"authenticated product smoke: {args.base_url.rstrip('/')}")
     login(client, admin_email, admin_password)
     print("login: ok")
+    workspace_id, workspace_slug = resolve_workspace(
+        client,
+        args.workspace_id,
+        args.workspace_slug,
+    )
+    print(f"workspace: {workspace_id} / {workspace_slug}")
 
     operation_id = create_operation(
         client,
-        args.workspace_id,
+        workspace_id,
         operation_name,
         args.internal_upstream,
     )
     print(f"operation created: {operation_id}")
-    publish_operation(client, args.workspace_id, operation_id)
+    publish_operation(client, workspace_id, operation_id)
     print("operation published: v1")
 
-    agent_id = create_agent(client, args.workspace_id, agent_slug)
-    bind_and_publish_agent(client, args.workspace_id, agent_id, operation_id, operation_name)
+    agent_id = create_agent(client, workspace_id, agent_slug)
+    bind_and_publish_agent(client, workspace_id, agent_id, operation_id, operation_name)
     print(f"agent published: {agent_id}")
 
-    api_key, key_id = create_agent_key(client, args.workspace_id, agent_id)
-    mcp_url = agent_mcp_url(args.base_url, args.workspace_slug, agent_slug)
+    api_key, key_id = create_agent_key(client, workspace_id, agent_id)
+    mcp_url = agent_mcp_url(args.base_url, workspace_slug, agent_slug)
     session_id = initialize_mcp_session(client, mcp_url, api_key)
     print("mcp initialized: ok")
 
@@ -420,7 +455,7 @@ def run(args: argparse.Namespace) -> None:
     if "error" in result:
         raise SmokeError(f"tools/call returned error: {result['error']}")
     print("tools/call: ok")
-    cleanup_smoke_assets(client, args.workspace_id, operation_id, agent_id, key_id)
+    cleanup_smoke_assets(client, workspace_id, operation_id, agent_id, key_id)
     print("authenticated product smoke completed")
 
 

tests/unit/test_authenticated_product_smoke.py

@@ -35,10 +35,40 @@ class AuthenticatedProductSmokeTests(unittest.TestCase):
         smoke = load_smoke_module()
 
         self.assertEqual(
-            smoke.agent_mcp_url("https://crank.example.com/", "default", "health-smoke"),
-            "https://crank.example.com/mcp/v1/default/health-smoke",
+            smoke.agent_mcp_url("https://crank.example.com/", "solo", "health-smoke"),
+            "https://crank.example.com/mcp/v1/solo/health-smoke",
         )
 
+    def test_resolve_workspace_prefers_authenticated_session_slug(self) -> None:
+        smoke = load_smoke_module()
+
+        class FakeClient:
+            def request_json(self, method, path):
+                self.request = (method, path)
+                return smoke.JsonResponse(
+                    status=200,
+                    headers={},
+                    body={
+                        "current_workspace_id": "ws_default",
+                        "memberships": [
+                            {
+                                "workspace": {
+                                    "id": "ws_default",
+                                    "slug": "solo",
+                                }
+                            }
+                        ],
+                    },
+                )
+
+        client = FakeClient()
+
+        self.assertEqual(
+            smoke.resolve_workspace(client, "ws_default", "default"),
+            ("ws_default", "solo"),
+        )
+        self.assertEqual(client.request, ("GET", "/api/auth/session"))
+
     def test_tool_call_payload_targets_health_tool(self) -> None:
         smoke = load_smoke_module()