fix(tasks): mirror parent_id to Project body on create when absent

src/albert/collections/tasks.py

@@ -78,6 +78,7 @@
     fetch_csv_table_rows,
     generate_adv_patch_payload,
     map_csv_headers_to_columns,
+    mirror_project_from_parent_id,
     resolve_attachment,
 )
 from albert.utils.worker_jobs import poll_worker_job
@@ -241,11 +242,14 @@ def create(self, *, task: PropertyTask | GeneralTask | BatchTask) -> BaseTask:
             The created task (a ``PropertyTask``, ``BatchTask``, or ``GeneralTask``),
             populated with its assigned Task ID.
         """
-        payload = [task.model_dump(mode="json", by_alias=True, exclude_none=True)]
+        payload = mirror_project_from_parent_id(
+            payload=task.model_dump(mode="json", by_alias=True, exclude_none=True),
+            parent_id=task.parent_id,
+        )
         url = f"{self.base_path}/multi?category={task.category.value}"
         if task.parent_id is not None:
             url = f"{url}&parentId={task.parent_id}"
-        response = self.session.post(url=url, json=payload)
+        response = self.session.post(url=url, json=[payload])
         task_data = response.json()[0]
         return TaskAdapter.validate_python(task_data)
 
@@ -300,7 +304,13 @@ def create_many(
         if len(parent_ids) != 1:
             raise AlbertException("All tasks in create_many must share the same parent_id.")
         task = tasks[0]
-        payload = [t.model_dump(mode="json", by_alias=True, exclude_none=True) for t in tasks]
+        payload = [
+            mirror_project_from_parent_id(
+                payload=t.model_dump(mode="json", by_alias=True, exclude_none=True),
+                parent_id=t.parent_id,
+            )
+            for t in tasks
+        ]
         url = f"{self.base_path}/multi?category={task.category.value}"
         if task.parent_id is not None:
             url = f"{url}&parentId={task.parent_id}"

src/albert/utils/tasks.py

@@ -25,6 +25,13 @@
 CSV_EXTENSIONS: set[str] = {"csv"}
 
 
+def mirror_project_from_parent_id(*, payload: dict, parent_id: str | None) -> dict:
+    """Mirror ``parent_id`` into the payload's ``Project`` link when it is a Project ID."""
+    if parent_id is not None and "Project" not in payload and parent_id.upper().startswith("PRO"):
+        payload["Project"] = {"id": parent_id}
+    return payload
+
+
 def build_property_payload(
     *,
     data_rows: Iterable[dict[str, dict]],

tests/integration/collections/test_tasks.py

@@ -137,6 +137,27 @@ def test_create_many_rejects_mixed_categories(client: Albert):
         client.tasks.create_many(tasks=[GeneralTask(name="a"), BatchTask(name="b")])
 
 
+def test_create_with_project_parent_id_links_project(
+    client: Albert, seed_prefix: str, seeded_locations, seeded_projects
+):
+    """Test that create links the task to the project when only parent_id is set."""
+    project = seeded_projects[0]
+    task = client.tasks.create(
+        task=GeneralTask(
+            name=f"{seed_prefix} - project parent link",
+            location=seeded_locations[0],
+            parent_id=project.id,
+        )
+    )
+    try:
+        fetched = client.tasks.get_by_id(id=task.id)
+        links = fetched.project if isinstance(fetched.project, list) else [fetched.project]
+        assert project.id in {link.id for link in links if link is not None}
+    finally:
+        with suppress(NotFoundError, BadRequestError):
+            client.tasks.delete(id=task.id)
+
+
 def test_delete_with_delay(client: Albert, seed_prefix: str, seeded_locations):
     """Test that delete accepts the optional delay parameter."""
     task = client.tasks.create(

tests/unit/utils/test_tasks.py

@@ -35,10 +35,60 @@
     generate_adv_patch_payload,
     is_metadata_item_list,
     map_csv_headers_to_columns,
+    mirror_project_from_parent_id,
     resolve_attachment,
 )
 from tests.unit.conftest import UNIT_BASE_URL
 
+# ---------------------------------------------------------------------------
+# mirror_project_from_parent_id
+# ---------------------------------------------------------------------------
+
+
+def test_mirror_project_from_parent_id_injects_project_for_project_parent():
+    """Test that a Project parent_id is mirrored into a payload missing Project."""
+    payload = {"name": "Task"}
+
+    result = mirror_project_from_parent_id(payload=payload, parent_id="PRO1")
+
+    assert result == {"name": "Task", "Project": {"id": "PRO1"}}
+
+
+def test_mirror_project_from_parent_id_preserves_existing_project():
+    """Test that an explicit Project in the payload is never overwritten."""
+    payload = {"name": "Task", "Project": {"id": "PRO9"}}
+
+    result = mirror_project_from_parent_id(payload=payload, parent_id="PRO1")
+
+    assert result == {"name": "Task", "Project": {"id": "PRO9"}}
+
+
+def test_mirror_project_from_parent_id_ignores_non_project_parent():
+    """Test that an Inventory parent_id is not mirrored into Project."""
+    payload = {"name": "Task"}
+
+    result = mirror_project_from_parent_id(payload=payload, parent_id="INV1")
+
+    assert result == {"name": "Task"}
+
+
+def test_mirror_project_from_parent_id_no_parent_leaves_payload_untouched():
+    """Test that a missing parent_id leaves the payload unchanged."""
+    payload = {"name": "Task"}
+
+    result = mirror_project_from_parent_id(payload=payload, parent_id=None)
+
+    assert result == {"name": "Task"}
+
+
+def test_mirror_project_from_parent_id_accepts_lowercase_prefix():
+    """Test that a lowercase project prefix is still recognized as a Project ID."""
+    payload = {"name": "Task"}
+
+    result = mirror_project_from_parent_id(payload=payload, parent_id="pro1")
+
+    assert result == {"name": "Task", "Project": {"id": "pro1"}}
+
 
 def _payload(offline_session, *, existing: GeneralTask, updated: GeneralTask):
     return generate_adv_patch_payload(