fix: allows_same_name_different_parent

backend/src/backend/scripts.py

@@ -371,16 +371,9 @@ async def create_script_record(
                 "parent directory not found",
             )
 
-    display_parent = parent
-    child_path = f"{display_parent}/{name}" if display_parent else name
-    relative_path = user_relative_path(context, child_path)
-    logger.debug(relative_path)
-
     # The Jupyter-side filename is the sanitized script name; extension and
     # path separators are enforced by safe_script_name(). The script_id ULID
-    # remains the Scripts PK only. We still do a Scripts-only conflict check
-    # on the user-supplied name so two scripts cannot claim the same display
-    # name within the same workspace.
+    # remains the Scripts PK only.
     user_id = context.user.user_id
     script_id = new_ulid()  # PK only — kept as ULID
     jupyter_basename = _jupyter_path(script_type, name)  # name is already passed through safe_script_name
@@ -398,17 +391,33 @@ async def create_script_record(
         jupyter_path = f"{user_id}/{parent_segment}/{jupyter_basename}"
     else:
         jupyter_path = f"{user_id}/{jupyter_basename}"
+
+    # StorageObjects relative_path mirrors the Jupyter layout under the
+    # workspace/user prefix. Compute it before the conflict check so we can
+    # scope duplicates by the full path, not just the display name.
+    relative_path = user_relative_path(
+        context, f"{parent}/{jupyter_basename}" if parent else jupyter_basename
+    )
+    logger.debug(relative_path)
+
     name_clash = await session.scalar(
-        select(Scripts.script_id).where(
+        select(Scripts.script_id)
+        .join(
+            StorageObjects,
+            StorageObjects.storage_object_id == Scripts.current_object_id,
+        )
+        .where(
             Scripts.workspace_id == context.workspace.workspace_id,
+            Scripts.owner_user_id == context.user.user_id,
             Scripts.script_name == name,
             Scripts.status == "active",
+            StorageObjects.relative_path == relative_path,
         )
     )
     if name_clash is not None:
         raise HTTPException(
             status.HTTP_409_CONFLICT,
-            "a script with the same name already exists in this workspace",
+            "a script with the same name already exists at this path",
         )
 
     # Push the file directly to the workspace's live Jupyter instance.
@@ -475,9 +484,6 @@ async def create_script_record(
     object_key = await _resolve_unique_object_key(session, object_key)
     object_key_hash = hashlib.sha256(object_key.encode("utf-8")).digest()
     bucket_name = settings.s3_workspace_bucket
-    relative_path = user_relative_path(
-        context, f"{parent}/{jupyter_basename}" if parent else jupyter_basename
-    )
     mime_type = mimetypes.guess_type(jupyter_basename)[0]
     storage_object = StorageObjects(
         storage_object_id=object_id,

backend/src/backend/services/storage.py

@@ -48,7 +48,7 @@ from common.storage.schemas import (
     DownloadUrlRequest,
     ServerObjectRequest,
 )
-from common.storage import build_storage_uri
+from common.storage import actual_bucket_name, build_storage_uri
 
 
 # ── shared low-level helpers (module-private) ────────────────────────────
@@ -590,7 +590,7 @@ async def soft_delete_object(
                 "storage_object_id": storage_object_id,
                 "object_status": item.object_status,
                 "trash_key": item.trash_key,
-                "trash_bucket": settings.s3_trash_bucket,
+                "trash_bucket": actual_bucket_name("trash"),
             }
         }
     if item.storage_backend == settings.storage_backend and item.bucket_name and item.object_key:
@@ -598,7 +598,7 @@ async def soft_delete_object(
         try:
             object_stores = request.app.state.object_stores
             data = await object_stores[item.bucket_name].get(item.object_key)
-            await object_stores[settings.s3_trash_bucket].put(trash_key, data)
+            await object_stores[actual_bucket_name("trash")].put(trash_key, data)
             await object_stores[item.bucket_name].delete(item.object_key)
         except Exception as exc:
             raise HTTPException(
@@ -614,6 +614,6 @@ async def soft_delete_object(
             "storage_object_id": storage_object_id,
             "object_status": item.object_status,
             "trash_key": item.trash_key,
-            "trash_bucket": settings.s3_trash_bucket,
+            "trash_bucket": actual_bucket_name("trash"),
         }
     }

backend/src/backend/storage_api.py

@@ -479,7 +479,7 @@ async def restore_object(
     try:
         # Cross-backend copy: get from trash, put back to source bucket.
         object_stores = request.app.state.object_stores
-        data = await object_stores[settings.s3_trash_bucket].get(item.trash_key)
+        data = await object_stores[actual_bucket_name("trash")].get(item.trash_key)
         await object_stores[item.bucket_name].put(item.object_key, data)
     except Exception as exc:
         raise HTTPException(
@@ -529,7 +529,7 @@ async def purge_trash_object(
         )
     if item.trash_key:
         try:
-            await request.app.state.object_stores[settings.s3_trash_bucket].delete(
+            await request.app.state.object_stores[actual_bucket_name("trash")].delete(
                 item.trash_key
             )
         except Exception as exc:

backend/tests/test_scripts.py

@@ -83,6 +83,7 @@ class _AsyncSessionMock:
         self.added: list[object] = []
         self.flush_order: list[str] = []
         self._refreshed: list[object] = []
+        self.scalar_results: list[Any] = []
 
     def add(self, obj: object) -> None:
         self.added.append(obj)
@@ -95,6 +96,8 @@ class _AsyncSessionMock:
         self._refreshed.append(obj)
 
     async def scalar(self, *_args, **_kwargs) -> None:
+        if self.scalar_results:
+            return self.scalar_results.pop(0)
         return None
 
 
@@ -222,6 +225,60 @@ async def test_create_script_record_allows_reupload_after_delete() -> None:
     ]
 
 
+@pytest.mark.asyncio
+async def test_create_script_record_allows_same_name_different_parent() -> None:
+    """Same filename in different parent directories of the same user
+    must coexist — they correspond to different Jupyter paths
+    (/user/foo.ipynb vs /user/test/foo.ipynb).
+    """
+    from backend.scripts import create_script_record
+
+    session = _AsyncSessionMock()
+    request = _make_request()
+    context = _make_context()
+
+    runtime_client = request.app.state.runtime_client
+    runtime_client.ensure_directory = AsyncMock(return_value=None)
+    runtime_client.create_notebook = AsyncMock(return_value={"name": "x.ipynb"})
+
+    s1, so1 = await create_script_record(
+        name="x.ipynb", script_type="notebook", content=b'{"cells":[]}',
+        visibility="private", parent_path=None,
+        request=request, context=context, session=session,
+    )
+
+    # Prime the mock to return a directory row for the upcoming parent lookup.
+    user_prefix = f"workspace/{context.user.user_id}"
+    session.scalar_results.append(
+        StorageObjects(
+            storage_object_id="01OBJ0000000000000000AB",
+            workspace_id=context.workspace.workspace_id,
+            owner_user_id=context.user.user_id,
+            object_type="directory",
+            usage_type="working_copy",
+            storage_backend=settings.storage_backend,
+            storage_uri="s3://bucket/key",
+            file_name="test",
+            created_by=context.user.user_id,
+            relative_path=f"{user_prefix}/test",
+        )
+    )
+
+    # Second upload with the same name but in a subdirectory must NOT 409.
+    s2, so2 = await create_script_record(
+        name="x.ipynb", script_type="notebook", content=b'{"cells":[]}',
+        visibility="private", parent_path="test",
+        request=request, context=context, session=session,
+    )
+
+    assert s1.script_id != s2.script_id
+    assert so1.storage_object_id != so2.storage_object_id
+    # Different relative_paths because parent_path differs.
+    assert so1.relative_path != so2.relative_path
+    assert so1.relative_path.endswith("x.ipynb")
+    assert so2.relative_path.endswith("test/x.ipynb")
+
+
 @pytest.mark.asyncio
 async def test_create_script_after_soft_delete_does_not_conflict() -> None:
     """Resurrection regression: with uk_storage_bucket_key_active being a