[CHORE]: Rename tasks/operators to attached functions and functions (#5733)

## Description of changes

As titled.
Operators -> Functions
Task -> Attached Functions
TaskRuns -> Invocations

CreateTask -> AttachFunction
RemoveTask -> DetachFunction

Will rename files in a followup diff.

- Improvements & Bug fixes
    - ...
- New functionality
    - ...

## Test plan

_How are these changes tested?_

- [ ] Tests pass locally with `pytest` for python, `yarn test` for js, `cargo test` for rust

## Migration plan

_Are there any migrations, or any forwards/backwards compatibility changes needed in order to make sure this change deploys reliably?_

## Observability plan

_What is the plan to instrument and monitor this change?_

## Documentation Changes

_Are all docstrings for user-facing APIs updated if required? Do we need to make documentation changes in the_ [_docs section](https://github.com/chroma-core/chroma/tree/main/docs/docs.trychroma.com)?_

Author: tanujnay112

chromadb/api/__init__.py

@@ -72,6 +72,7 @@
 from chromadb.config import Component, Settings
 from chromadb.types import Database, Tenant, Collection as CollectionModel
 from chromadb.api.models.Collection import Collection
+from chromadb.api.models.AttachedFunction import AttachedFunction
 
 # Re-export the async version
 from chromadb.api.async_api import (  # noqa: F401
@@ -814,46 +815,44 @@ def _delete(
         pass
 
     @abstractmethod
-    def create_task(
+    def attach_function(
         self,
-        task_name: str,
-        operator_name: str,
+        function_id: str,
+        name: str,
         input_collection_id: UUID,
-        output_collection_name: str,
+        output_collection: str,
         params: Optional[Dict[str, Any]] = None,
         tenant: str = DEFAULT_TENANT,
         database: str = DEFAULT_DATABASE,
-    ) -> tuple[bool, str]:
-        """Create a recurring task on a collection.
+    ) -> "AttachedFunction":
+        """Attach a function to a collection.
 
         Args:
-            task_name: Unique name for this task instance
-            operator_name: Built-in operator name (e.g., 'record_counter')
-            input_collection_id: Source collection that triggers the task
-            output_collection_name: Target collection where task output is stored
-            params: Optional dictionary with operator-specific parameters
+            function_id: Built-in function identifier
+            name: Unique name for this attached function
+            input_collection_id: Source collection that triggers the function
+            output_collection: Target collection where function output is stored
+            params: Optional dictionary with function-specific parameters
             tenant: The tenant name
             database: The database name
 
         Returns:
-            tuple: (success: bool, task_id: str)
+            AttachedFunction: Object representing the attached function
         """
         pass
 
     @abstractmethod
-    def remove_task(
+    def detach_function(
         self,
-        task_name: str,
-        input_collection_id: UUID,
+        attached_function_id: UUID,
         delete_output: bool = False,
         tenant: str = DEFAULT_TENANT,
         database: str = DEFAULT_DATABASE,
     ) -> bool:
-        """Delete a task and prevent any further runs.
+        """Detach a function and prevent any further runs.
 
         Args:
-            task_name: Name of the task to remove
-            input_collection_id: Id of the input collection the task is registered on
+            attached_function_id: ID of the attached function to remove
             delete_output: Whether to also delete the output collection
             tenant: The tenant name
             database: The database name

chromadb/api/fastapi.py

@@ -7,6 +7,8 @@
 import urllib.parse
 from overrides import override
 
+from chromadb.api.models.AttachedFunction import AttachedFunction
+
 from chromadb.api.collection_configuration import (
     CreateCollectionConfiguration,
     UpdateCollectionConfiguration,
@@ -748,47 +750,56 @@ def get_max_batch_size(self) -> int:
         max_batch_size = cast(int, pre_flight_checks.get("max_batch_size", -1))
         return max_batch_size
 
-    @trace_method("FastAPI.create_task", OpenTelemetryGranularity.ALL)
+    @trace_method("FastAPI.attach_function", OpenTelemetryGranularity.ALL)
     @override
-    def create_task(
+    def attach_function(
         self,
-        task_name: str,
-        operator_name: str,
+        function_id: str,
+        name: str,
         input_collection_id: UUID,
-        output_collection_name: str,
+        output_collection: str,
         params: Optional[Dict[str, Any]] = None,
         tenant: str = DEFAULT_TENANT,
         database: str = DEFAULT_DATABASE,
-    ) -> tuple[bool, str]:
-        """Register a recurring task on a collection."""
+    ) -> "AttachedFunction":
+        """Attach a function to a collection."""
         resp_json = self._make_request(
             "post",
-            f"/tenants/{tenant}/databases/{database}/collections/{input_collection_id}/tasks/create",
+            f"/tenants/{tenant}/databases/{database}/collections/{input_collection_id}/functions/attach",
             json={
-                "task_name": task_name,
-                "operator_name": operator_name,
-                "output_collection_name": output_collection_name,
+                "name": name,
+                "function_id": function_id,
+                "output_collection": output_collection,
                 "params": params,
             },
         )
-        return cast(bool, resp_json["success"]), cast(str, resp_json["task_id"])
 
-    @trace_method("FastAPI.remove_task", OpenTelemetryGranularity.ALL)
+        return AttachedFunction(
+            client=self,
+            id=UUID(resp_json["attached_function"]["id"]),
+            name=resp_json["attached_function"]["name"],
+            function_id=resp_json["attached_function"]["function_id"],
+            input_collection_id=input_collection_id,
+            output_collection=output_collection,
+            params=params,
+            tenant=tenant,
+            database=database,
+        )
+
+    @trace_method("FastAPI.detach_function", OpenTelemetryGranularity.ALL)
     @override
-    def remove_task(
+    def detach_function(
         self,
-        task_name: str,
-        input_collection_id: UUID,
+        attached_function_id: UUID,
         delete_output: bool = False,
         tenant: str = DEFAULT_TENANT,
         database: str = DEFAULT_DATABASE,
     ) -> bool:
-        """Delete a task and prevent any further runs."""
+        """Detach a function and prevent any further runs."""
         resp_json = self._make_request(
             "post",
-            f"/tenants/{tenant}/databases/{database}/collections/{input_collection_id}/tasks/delete",
+            f"/tenants/{tenant}/databases/{database}/attached_functions/{attached_function_id}/detach",
             json={
-                "task_name": task_name,
                 "delete_output": delete_output,
             },
         )

chromadb/api/models/AttachedFunction.py

@@ -0,0 +1,101 @@
+from typing import TYPE_CHECKING, Optional, Dict, Any
+from uuid import UUID
+
+if TYPE_CHECKING:
+    from chromadb.api import ServerAPI  # noqa: F401
+
+
+class AttachedFunction:
+    """Represents a function attached to a collection."""
+
+    def __init__(
+        self,
+        client: "ServerAPI",
+        id: UUID,
+        name: str,
+        function_id: str,
+        input_collection_id: UUID,
+        output_collection: str,
+        params: Optional[Dict[str, Any]],
+        tenant: str,
+        database: str,
+    ):
+        """Initialize an AttachedFunction.
+
+        Args:
+            client: The API client
+            id: Unique identifier for this attached function
+            name: Name of this attached function instance
+            function_id: The function identifier (e.g., "record_counter")
+            input_collection_id: ID of the input collection
+            output_collection: Name of the output collection
+            params: Function-specific parameters
+            tenant: The tenant name
+            database: The database name
+        """
+        self._client = client
+        self._id = id
+        self._name = name
+        self._function_id = function_id
+        self._input_collection_id = input_collection_id
+        self._output_collection = output_collection
+        self._params = params
+        self._tenant = tenant
+        self._database = database
+
+    @property
+    def id(self) -> UUID:
+        """The unique identifier of this attached function."""
+        return self._id
+
+    @property
+    def name(self) -> str:
+        """The name of this attached function instance."""
+        return self._name
+
+    @property
+    def function_id(self) -> str:
+        """The function identifier."""
+        return self._function_id
+
+    @property
+    def input_collection_id(self) -> UUID:
+        """The ID of the input collection."""
+        return self._input_collection_id
+
+    @property
+    def output_collection(self) -> str:
+        """The name of the output collection."""
+        return self._output_collection
+
+    @property
+    def params(self) -> Optional[Dict[str, Any]]:
+        """The function parameters."""
+        return self._params
+
+    def detach(self, delete_output_collection: bool = False) -> bool:
+        """Detach this function and prevent any further runs.
+
+        Args:
+            delete_output_collection: Whether to also delete the output collection. Defaults to False.
+
+        Returns:
+            bool: True if successful
+
+        Example:
+            >>> success = attached_fn.detach(delete_output_collection=True)
+        """
+        return self._client.detach_function(
+            attached_function_id=self._id,
+            delete_output=delete_output_collection,
+            tenant=self._tenant,
+            database=self._database,
+        )
+
+    def __repr__(self) -> str:
+        return (
+            f"AttachedFunction(id={self._id}, name='{self._name}', "
+            f"function_id='{self._function_id}', "
+            f"input_collection_id={self._input_collection_id}, "
+            f"output_collection='{self._output_collection}')"
+        )

chromadb/api/models/Collection.py

@@ -25,6 +25,9 @@
 
 import logging
 
+if TYPE_CHECKING:
+    from chromadb.api.models.AttachedFunction import AttachedFunction
+
 logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
@@ -368,7 +371,7 @@ def search(
 
         return self._client._search(
             collection_id=self.id,
-            searches=cast(List[Search], embedded_searches),
+            searches=searches_list,
             tenant=self.tenant,
             database=self.database,
         )
@@ -495,63 +498,38 @@ def delete(
             database=self.database,
         )
 
-    def create_task(
+    def attach_function(
         self,
-        task_name: str,
-        operator_name: str,
-        output_collection_name: str,
+        function_id: str,
+        name: str,
+        output_collection: str,
         params: Optional[Dict[str, Any]] = None,
-    ) -> tuple[bool, str]:
-        """Create a recurring task that processes this collection.
+    ) -> "AttachedFunction":
+        """Attach a function to this collection.
 
         Args:
-            task_name: Unique name for this task instance
-            operator_name: Built-in operator name (e.g., "record_counter")
-            output_collection_name: Name of the collection where task output will be stored
-            params: Optional dictionary with operator-specific parameters
+            function_id: Built-in function identifier (e.g., "record_counter")
+            name: Unique name for this attached function
+            output_collection: Name of the collection where function output will be stored
+            params: Optional dictionary with function-specific parameters
 
         Returns:
-            tuple: (success: bool, task_id: str)
+            AttachedFunction: Object representing the attached function
 
         Example:
-            >>> success, task_id = collection.create_task(
-            ...     task_name="count_docs",
-            ...     operator_name="record_counter",
-            ...     output_collection_name="doc_counts",
+            >>> attached_fn = collection.attach_function(
+            ...     function_id="record_counter",
+            ...     name="mycoll_stats_fn",
+            ...     output_collection="mycoll_stats",
             ...     params={"threshold": 100}
             ... )
         """
-        return self._client.create_task(
-            task_name=task_name,
-            operator_name=operator_name,
+        return self._client.attach_function(
+            function_id=function_id,
+            name=name,
             input_collection_id=self.id,
-            output_collection_name=output_collection_name,
+            output_collection=output_collection,
             params=params,
             tenant=self.tenant,
             database=self.database,
         )
-
-    def remove_task(
-        self,
-        task_name: str,
-        delete_output: bool = False,
-    ) -> bool:
-        """Delete a task and prevent any further runs.
-
-        Args:
-            task_name: Name of the task to remove
-            delete_output: Whether to also delete the output collection. Defaults to False.
-
-        Returns:
-            bool: True if successful
-
-        Example:
-            >>> success = collection.remove_task("count_docs", delete_output=True)
-        """
-        return self._client.remove_task(
-            task_name=task_name,
-            input_collection_id=self.id,
-            delete_output=delete_output,
-            tenant=self.tenant,
-            database=self.database,
-        )