dmux · Oct 29, 2026 · Oct 26, 2026 · Oct 26, 2026 · Oct 28, 2026 · Oct 29, 2026
diff --git a/src/minilake/services/workspace.py b/src/minilake/services/workspace.py
@@ -76,6 +76,18 @@ def _next_id() -> int:
     return obj_id
 
 
+def _file_metadata(normalized: str, object_type: ObjectType) -> dict[str, Any]:
+    """Build complete file metadata before writing the file's contents."""
+    existing = _state["objects"].get(normalized, {})
+    now_ms = int(time.time() * 1000)
+    return {
+        "object_type": object_type.value,
+        "object_id": existing.get("object_id") or _next_id(),
+        "created_at": existing.get("created_at") or now_ms,
+        "modified_at": now_ms,
+    }
+
+
 def _directory_object_id(normalized: str) -> int:
     """Get (lazily assigning and caching) a directory's object_id.
 
@@ -155,22 +167,23 @@ async def import_object(req: ImportWorkspaceRequest) -> dict:
             status_code=400,
         )
 
+    if file_path.is_dir():
+        raise DatabricksError(
+            error_code="INVALID_REQUEST",
+            message=f"Path '{normalized}' is a directory",
+            status_code=400,
+        )
+
     file_path.parent.mkdir(parents=True, exist_ok=True)
     try:
         content_bytes = base64.b64decode(req.content, validate=True) if req.content else b""
     except Exception as e:
         raise DatabricksError(error_code="INVALID_REQUEST", message=f"Invalid base64 content: {e}", status_code=400)
-    file_path.write_bytes(content_bytes)
 
-    now_ms = int(time.time() * 1000)
-    existing = _state["objects"].get(normalized)
-    _state["objects"][normalized] = {
-        "object_type": ObjectType.NOTEBOOK.value,
-        "object_id": existing["object_id"] if existing else _next_id(),
-        "language": Language.PYTHON.value,
-        "created_at": existing["created_at"] if existing else now_ms,
-        "modified_at": now_ms,
-    }
+    metadata = _file_metadata(normalized, ObjectType.NOTEBOOK)
+    metadata["language"] = Language.PYTHON.value
+    file_path.write_bytes(content_bytes)
+    _state["objects"][normalized] = metadata
 
     logger.info(f"Imported workspace object: {normalized}")
     return {}
@@ -193,6 +206,13 @@ async def import_file(
     normalized = _normalize("/" + path)
     file_path = _resolve(normalized)
 
+    if file_path.is_dir():
+        raise DatabricksError(
+            error_code="INVALID_REQUEST",
+            message=f"Path '{normalized}' is a directory",
+            status_code=400,
+        )
+
     if file_path.exists() and not file_path.is_dir() and not overwrite:
         raise DatabricksError(
             error_code="RESOURCE_ALREADY_EXISTS",
@@ -202,16 +222,9 @@ async def import_file(
 
     file_path.parent.mkdir(parents=True, exist_ok=True)
     body = await request.body()
+    metadata = _file_metadata(normalized, ObjectType.FILE)
     file_path.write_bytes(body)
-
-    now_ms = int(time.time() * 1000)
-    existing = _state["objects"].get(normalized)
-    _state["objects"][normalized] = {
-        "object_type": ObjectType.FILE.value,
-        "object_id": existing["object_id"] if existing else _next_id(),
-        "created_at": existing["created_at"] if existing else now_ms,
-        "modified_at": now_ms,
-    }
+    _state["objects"][normalized] = metadata
 
     logger.info(f"Imported workspace file: {normalized}")
     return {}
@@ -385,6 +398,7 @@ async def delete_object(req: DeleteWorkspaceRequest) -> dict:
                 status_code=400,
             )
         shutil.rmtree(target)
+        _state["objects"].pop(normalized, None)
         prefix = normalized.rstrip("/") + "/"
         for key in list(_state["objects"].keys()):
             if key.startswith(prefix):

diff --git a/tests/test_workspace.py b/tests/test_workspace.py
@@ -4,6 +4,7 @@
 
 import pytest
 from databricks.sdk import WorkspaceClient
+from databricks.sdk.errors import DatabricksError
 from databricks.sdk.service.workspace import ImportFormat, Language
 
 SAMPLE_SCRIPT = b"print('hello from minilake workspace test')\n"
@@ -82,6 +83,50 @@ def test_workspace_mkdirs_and_delete(workspace_client: WorkspaceClient):
     print("✓ Workspace mkdirs/delete works for real")
 
 
+@pytest.mark.workflow
+def test_workspace_deleted_directory_can_be_reused_for_notebook(workspace_client: WorkspaceClient):
+    path = "/Shared/replaced_directory.py"
+    workspace_client.workspace.mkdirs(path=path)
+    old_directory = workspace_client.workspace.get_status(path=path)
+    workspace_client.workspace.delete(path=path, recursive=True)
+
+    workspace_client.workspace.import_(
+        path=path,
+        content=base64.b64encode(SAMPLE_SCRIPT).decode("ascii"),
+        format=ImportFormat.SOURCE,
+        language=Language.PYTHON,
+        overwrite=True,
+    )
+
+    status = workspace_client.workspace.get_status(path=path)
+    assert status.object_type.value == "NOTEBOOK"
+    assert status.object_id != old_directory.object_id
+    assert status.created_at is not None
+    assert base64.b64decode(workspace_client.workspace.export(path=path).content) == SAMPLE_SCRIPT
+
+
+@pytest.mark.error
+def test_workspace_notebook_import_does_not_replace_directory(workspace_client: WorkspaceClient):
+    path = "/Shared/not_a_notebook_directory"
+    workspace_client.workspace.mkdirs(path=path)
+    before = workspace_client.workspace.get_status(path=path)
+
+    with pytest.raises(DatabricksError) as error:
+        workspace_client.workspace.import_(
+            path=path,
+            content=base64.b64encode(SAMPLE_SCRIPT).decode("ascii"),
+            format=ImportFormat.SOURCE,
+            language=Language.PYTHON,
+            overwrite=False,
+        )
+
+    assert error.value.error_code == "RESOURCE_ALREADY_EXISTS"
+    after = workspace_client.workspace.get_status(path=path)
+    assert after.object_type.value == "DIRECTORY"
+    assert after.object_id == before.object_id
+    assert list(workspace_client.workspace.list(path=path)) == []
+
+
 @pytest.mark.error
 def test_workspace_export_nonexistent_fails(workspace_client: WorkspaceClient):
     """Test: Exporting a nonexistent path raises an error."""

diff --git a/tests/test_workspace_files.py b/tests/test_workspace_files.py
@@ -57,6 +57,38 @@ def test_imported_file_reports_object_type_file(workspace_client: WorkspaceClien
     print("✓ synced file reports object_type=FILE")
 
 
+@pytest.mark.workflow
+def test_workspace_file_deleted_directory_can_be_reused(workspace_client: WorkspaceClient):
+    path = "/Shared/replaced_directory.txt"
+    content = b"file after directory deletion"
+    workspace_client.workspace.mkdirs(path=path)
+    old_directory = workspace_client.workspace.get_status(path=path)
+    workspace_client.workspace.delete(path=path, recursive=True)
+
+    _import_file(workspace_client, path, content)
+
+    status = workspace_client.workspace.get_status(path=path)
+    assert status.object_type.value == "FILE"
+    assert status.object_id != old_directory.object_id
+    assert status.created_at is not None
+    assert _read_file(workspace_client, path) == content
+
+
+@pytest.mark.error
+def test_workspace_file_import_does_not_replace_directory(workspace_client: WorkspaceClient):
+    path = "/Shared/not_a_file_directory"
+    workspace_client.workspace.mkdirs(path=path)
+    before = workspace_client.workspace.get_status(path=path)
+
+    with pytest.raises(Exception):
+        _import_file(workspace_client, path, b"file content")
+
+    after = workspace_client.workspace.get_status(path=path)
+    assert after.object_type.value == "DIRECTORY"
+    assert after.object_id == before.object_id
+    assert list(workspace_client.workspace.list(path=path)) == []
+
+
 @pytest.mark.error
 def test_import_file_overwrite_false_conflicts(workspace_client: WorkspaceClient):
     """Re-importing without overwrite raises (RESOURCE_ALREADY_EXISTS / 400)."""