feat(tasks): enforce unassigned tasks start Unclaimed on create

src/albert/collections/tasks.py

@@ -80,6 +80,7 @@
     map_csv_headers_to_columns,
     mirror_project_from_parent_id,
     resolve_attachment,
+    resolve_unassigned_task_state,
 )
 from albert.utils.worker_jobs import poll_worker_job
 
@@ -223,6 +224,15 @@ def create(self, *, task: PropertyTask | GeneralTask | BatchTask) -> BaseTask:
         For a PropertyTask, set ``parent_id`` to the parent Project ID. Blocks are
         added separately with [`add_block`][albert.collections.tasks.TaskCollection.add_block] after creation.
 
+        Lifecycle state on create is coupled to the assignee:
+
+        - No ``assigned_to``: the task starts ``Unclaimed``. A missing ``state``
+          is defaulted for you, and ``Not Started`` / ``In Progress`` are rejected,
+          since a task cannot be in progress with no one assigned. Explicit
+          terminal states (``Completed``, ``Closed``, ``Cancelled``) are preserved
+          for ingestion flows.
+        - With ``assigned_to``: the task starts ``Not Started``.
+
         !!! example
             ```python
             from albert.resources.tasks import GeneralTask
@@ -241,11 +251,19 @@ def create(self, *, task: PropertyTask | GeneralTask | BatchTask) -> BaseTask:
         BaseTask
             The created task (a ``PropertyTask``, ``BatchTask``, or ``GeneralTask``),
             populated with its assigned Task ID.
+
+        Raises
+        ------
+        AlbertException
+            If ``task`` has no ``assigned_to`` but ``state`` is ``Not Started`` or
+            ``In Progress``.
         """
         payload = mirror_project_from_parent_id(
             payload=task.model_dump(mode="json", by_alias=True, exclude_none=True),
             parent_id=task.parent_id,
         )
+        if task.assigned_to is None:
+            payload["state"] = resolve_unassigned_task_state(state=task.state).value
         url = f"{self.base_path}/multi?category={task.category.value}"
         if task.parent_id is not None:
             url = f"{url}&parentId={task.parent_id}"
@@ -262,6 +280,12 @@ def create_many(
         All tasks must share the same category (mixing task types is not
         supported) and, when set, the same ``parent_id``.
 
+        Lifecycle state on create follows the same rule as
+        [`create`][albert.collections.tasks.TaskCollection.create]: a task without
+        ``assigned_to`` starts ``Unclaimed`` (a missing ``state`` is defaulted, and
+        ``Not Started`` / ``In Progress`` are rejected), while a task with an
+        assignee starts ``Not Started``.
+
         !!! example
             ```python
             from albert.resources.tasks import GeneralTask
@@ -292,8 +316,9 @@ def create_many(
         Raises
         ------
         AlbertException
-            If ``tasks`` is empty or if items have conflicting categories or
-            parent IDs.
+            If ``tasks`` is empty, if items have conflicting categories or
+            parent IDs, or if any task has no ``assigned_to`` but ``state`` is
+            ``Not Started`` or ``In Progress``.
         """
         if not tasks:
             raise AlbertException("tasks must include at least one task.")
@@ -311,6 +336,9 @@ def create_many(
             )
             for t in tasks
         ]
+        for t, item in zip(tasks, payload, strict=True):
+            if t.assigned_to is None:
+                item["state"] = resolve_unassigned_task_state(state=t.state).value
         url = f"{self.base_path}/multi?category={task.category.value}"
         if task.parent_id is not None:
             url = f"{url}&parentId={task.parent_id}"
@@ -2089,6 +2117,10 @@ def update(self, *, task: BaseTask) -> BaseTask:
         The following fields can be updated: ``assigned_to``, ``due_date``,
         ``inventory_information``, ``metadata``, ``name``, ``priority``, ``project``,
         ``state``, ``tags``.
+
+        ``state`` and ``assigned_to`` are coupled: a task cannot be ``In Progress``
+        (or any state beyond ``Not Started``) while unassigned, so moving an
+        unclaimed task forward also assigns it to the calling user.
         """
         existing = self.get_by_id(id=task.id)
         patch_payload = generate_adv_patch_payload(

src/albert/resources/tasks.py

@@ -203,6 +203,12 @@ class TaskState(str, Enum):
     The state generally advances as work progresses: an unclaimed task is
     picked up, started, finished, and eventually closed (or cancelled).
 
+    State is coupled to the assignee: a task without an assignee is always
+    ``Unclaimed``, and a task cannot be ``Not Started`` or ``In Progress``
+    with no one assigned. Creating a task without an assignee defaults its
+    state to ``Unclaimed``; see
+    [`create`][albert.collections.tasks.TaskCollection.create].
+
     Attributes
     ----------
     UNCLAIMED : str
@@ -450,7 +456,7 @@ class BaseTask(BaseTaggedResource):
     """Overall result summary for the task."""
 
     state: TaskState | None = Field(default=None)
-    """Current lifecycle state of the task."""
+    """Current lifecycle state of the task. A task without an ``assigned_to`` is always ``Unclaimed``; ``Not Started`` and ``In Progress`` require an assignee. On create, a missing state defaults to ``Unclaimed`` when the task has no assignee."""
 
     project: SerializeAsEntityLink[Project] | list[SerializeAsEntityLink[Project]] | None = Field(
         default=None, alias="Project"
@@ -460,7 +466,7 @@ class BaseTask(BaseTaggedResource):
     assigned_to: (
         SerializeAsEntityLinkWithName[User] | SerializeAsEntityLinkWithName[Team] | None
     ) = Field(default=None, alias="AssignedTo")
-    """The user or team responsible for the task."""
+    """The user or team responsible for the task. When ``None``, the task is unassigned and its ``state`` must be ``Unclaimed`` (terminal states such as ``Completed`` are allowed on create for ingestion flows)."""
 
     page_state: PageState | None = Field(
         alias="PageState",

src/albert/utils/tasks.py

@@ -9,6 +9,7 @@
 from albert.core.shared.identifiers import DataTemplateId
 from albert.core.shared.models.base import EntityLink, EntityLinkWithName
 from albert.core.shared.models.patch import PatchOperation
+from albert.exceptions import AlbertException
 from albert.resources.data_templates import DataColumnValue
 from albert.resources.property_data import TaskDataColumn, TaskPropertyCreate
 from albert.resources.tasks import (
@@ -20,6 +21,7 @@
     TaskMetadataDataTemplate,
     TaskMetadataWorkflow,
     TaskPatchPayload,
+    TaskState,
 )
 
 CSV_EXTENSIONS: set[str] = {"csv"}
@@ -32,6 +34,25 @@ def mirror_project_from_parent_id(*, payload: dict, parent_id: str | None) -> di
     return payload
 
 
+def resolve_unassigned_task_state(*, state: TaskState | None) -> TaskState:
+    """Resolve the create-time lifecycle state for a task that has no assignee.
+
+    An unassigned task must start Unclaimed. A missing state defaults to
+    ``TaskState.UNCLAIMED`` and ownership states (``NOT_STARTED``,
+    ``IN_PROGRESS``) are rejected, since a task cannot be claimed or in
+    progress with no one assigned. Explicit terminal states (``COMPLETED``,
+    ``CLOSED``, ``CANCELLED``) pass through for ingestion flows.
+    """
+    if state is None:
+        return TaskState.UNCLAIMED
+    if state in (TaskState.NOT_STARTED, TaskState.IN_PROGRESS):
+        raise AlbertException(
+            f"Task state '{state.value}' requires an assignee. Set assigned_to, or leave "
+            "state unset to create the task as 'Unclaimed'."
+        )
+    return state
+
+
 def build_property_payload(
     *,
     data_rows: Iterable[dict[str, dict]],

tests/unit/utils/test_tasks.py

@@ -16,6 +16,7 @@
 from albert.collections.tasks import TaskCollection
 from albert.core.shared.models.base import EntityLink, EntityLinkWithName
 from albert.core.shared.models.patch import PatchOperation
+from albert.exceptions import AlbertException
 from albert.resources.attachments import AttachmentMetadata
 from albert.resources.data_templates import CurveDataEntityLink, DataColumnValue
 from albert.resources.notes import Note, NoteAttachmentEntityLink
@@ -25,6 +26,7 @@
     GeneralTask,
     PropertyTask,
     TaskInventoryInformation,
+    TaskState,
 )
 from albert.utils.tasks import (
     build_property_payload,
@@ -37,6 +39,7 @@
     map_csv_headers_to_columns,
     mirror_project_from_parent_id,
     resolve_attachment,
+    resolve_unassigned_task_state,
 )
 from tests.unit.conftest import UNIT_BASE_URL
 
@@ -90,6 +93,43 @@ def test_mirror_project_from_parent_id_accepts_lowercase_prefix():
     assert result == {"name": "Task", "Project": {"id": "pro1"}}
 
 
+# ---------------------------------------------------------------------------
+# resolve_unassigned_task_state
+# ---------------------------------------------------------------------------
+
+
+def test_resolve_unassigned_task_state_defaults_missing_state_to_unclaimed():
+    """Test that an absent state resolves to Unclaimed."""
+    assert resolve_unassigned_task_state(state=None) is TaskState.UNCLAIMED
+
+
+@pytest.mark.parametrize(
+    "state",
+    [
+        pytest.param(TaskState.NOT_STARTED, id="not-started"),
+        pytest.param(TaskState.IN_PROGRESS, id="in-progress"),
+    ],
+)
+def test_resolve_unassigned_task_state_rejects_ownership_states(state):
+    """Test that ownership states are rejected without an assignee."""
+    with pytest.raises(AlbertException, match="requires an assignee"):
+        resolve_unassigned_task_state(state=state)
+
+
+@pytest.mark.parametrize(
+    "state",
+    [
+        pytest.param(TaskState.UNCLAIMED, id="unclaimed"),
+        pytest.param(TaskState.COMPLETED, id="completed"),
+        pytest.param(TaskState.CLOSED, id="closed"),
+        pytest.param(TaskState.CANCELLED, id="cancelled"),
+    ],
+)
+def test_resolve_unassigned_task_state_preserves_allowed_explicit_states(state):
+    """Test that explicit Unclaimed and terminal states pass through unchanged."""
+    assert resolve_unassigned_task_state(state=state) is state
+
+
 def _payload(offline_session, *, existing: GeneralTask, updated: GeneralTask):
     return generate_adv_patch_payload(
         collection=TaskCollection(session=offline_session),