Debug a Project Assignment Codebase

Task Assignment Engine

Scale AI routes large volumes of data labeling tasks to qualified annotators based on project requirements, annotator skill profiles, and concurrency limits. You are tasked with implementing (or refactoring) the core ProjectAssignmentEngine class.

Requirements

Implement the ProjectAssignmentEngine class with the following methods:

1. register_annotator(annotator_id: str, skills: list[str], max_concurrent_tasks: int) -> None

   • Adds a new annotator or updates an existing annotator's skill list and capacity. Active tasks assigned to an existing annotator must remain active.

2. add_project(project_id: str, required_skills: list[str], priority: int) -> None

   • Registers a project. Higher priority values take precedence (priority=10 is higher priority than priority=1).

3. add_tasks(project_id: str, task_ids: list[str]) -> None

   • Appends unassigned task IDs to the given project. Tasks within the same project are assigned in FIFO order.

4. assign_next() -> list[tuple[str, str, str]]

   • Finds available capacity across annotators and assigns pending tasks.
   • Only projects with pending tasks are considered, processed in descending order of project priority. For ties in priority, use project insertion order.
   • A task can only be assigned to an annotator who possesses all of the project's required_skills and currently has active tasks strictly less than max_concurrent_tasks.
   • If multiple annotators are eligible for a task, assign to the one with the fewest active tasks. If there is still a tie, assign to the annotator whose ID is lexicographically smallest.
   • Returns a list of (task_id, project_id, annotator_id) tuples representing all new assignments made during this invocation.

5. complete_task(task_id: str) -> bool

   • Marks an active task as completed and frees up one concurrency slot for the assigned annotator.
   • Returns True if the task was currently active and successfully completed; returns False if task_id is unknown or not currently active.

Constraints

• Number of total tasks <= 10^5.
• Number of annotators <= 500.
• Number of projects <= 1,000.
• Length of skill sets <= 20 per annotator/project.
• All IDs are non-empty alphanumeric strings.

Example

engine = ProjectAssignmentEngine()
engine.register_annotator("ann_1", ["nlp", "audio"], max_concurrent_tasks=2)
engine.register_annotator("ann_2", ["nlp"], max_concurrent_tasks=1)

engine.add_project("proj_a", required_skills=["nlp"], priority=5)
engine.add_tasks("proj_a", ["t1", "t2"])

engine.add_project("proj_b", required_skills=["audio"], priority=10)
engine.add_tasks("proj_b", ["t3"])

# proj_b has higher priority (10 > 5), but requires 'audio', which only ann_1 has.
# proj_a requires 'nlp', which both ann_1 and ann_2 have.
assignments = engine.assign_next()
# Expected assignments:
# 1. t3 (proj_b) -> ann_1 (ann_1 load becomes 1)
# 2. t1 (proj_a) -> ann_2 (ann_2 load is 0, ann_1 load is 1; ann_2 has fewer active tasks)
# 3. t2 (proj_a) -> ann_1 (ann_2 is at capacity 1/1, ann_1 has capacity 1/2 left)
# Result: [("t3", "proj_b", "ann_1"), ("t1", "proj_a", "ann_2"), ("t2", "proj_a", "ann_1")]

engine.complete_task("t1")  # Returns True, ann_2 now has 0 active tasks
engine.complete_task("non_existent")  # Returns False