# Monitor Task Stop

## Status Transitions

```mermaid
stateDiagram-v2
    [*] --> pending : queue_monitor_task()

    pending --> in_processing : worker picks up
    pending --> awaiting_processing : stop_monitor_task_execution()

    in_processing --> complete : success
    in_processing --> error : failure
    in_processing --> awaiting_processing : stop_monitor_task_execution() (LLM classification only)

    awaiting_processing --> pending : retry_monitor_task() (new celery_id)
    complete --> pending : retry_monitor_task() (new celery_id)
    error --> pending : retry_monitor_task() (new celery_id)
```

## How stopping works

Stopping is triggered via `POST /tasks/{task_id}/stop`. The behavior depends on the task's current status:

- **Pending tasks** (any type): revoked from the Celery queue so they never start.
  Staged resources are reset immediately since the task will never run.
- **In-processing tasks** (LLM classification only): uses cooperative cancellation. A Redis flag is set for the task, and each worker thread checks this flag before making an LLM call. When it detects the stop, it skips remaining LLM calls, writes only the results for fields that were already classified, and resets unclassified resources back to their previous state. If Redis is unavailable, the check is skipped and classification continues normally. See the [Cooperative Cancellation](#cooperative-cancellation-in-processing-llm-classification-tasks) diagram below.
- **In-processing tasks** (other types): not yet supported. Returns an error.

If `include_group=true` is passed and the task has a `group_id`, all sibling tasks in the same group are also stopped.

## Stop Request Flow

```mermaid
flowchart TD
    A["POST /tasks/{task_id}/stop?include_group=true"] --> B{Task found?}
    B -- No --> B1[404 Not Found]
    B -- Yes --> D["Collect tasks to stop:\ntarget task \n+ siblings with same group_id\n(if include_group=true)"]
    D --> E{Any stoppable tasks?}
    E -- No --> E1[400 No stoppable tasks]
    E -- Yes --> F["For each task_to_stop:\nstop_monitor_task_execution(db, task)"]
    F --> G{Task status?}
    G -- pending --> H["revoke(celery_id)\nremoves from queue\nreset staged resources immediately"]
    G -- "in_processing\n(LLM classification only)" --> I["Update status only — cooperative\ncancellation handles the rest"]
    G -- "in_processing\n(other types)" --> J["ValueError:\nnot yet supported"]
    H & I --> K["Update status → awaiting_processing"]
    K --> L["Return List[MonitorTaskResponse]"]
```

## Cooperative Cancellation (in-processing LLM classification tasks)

```mermaid
sequenceDiagram
    participant API as API (stop request)
    participant Redis as Redis
    participant DB as Database
    participant Worker as Celery Worker
    participant Pool as ThreadPoolExecutor
    participant LLM as LLM Provider

    Note over Worker: Task is classifying fields (1 LLM call per field)
    Pool->>LLM: Call LLM (field 1)
    Pool->>LLM: Call LLM (field 2)
    Note over Pool: ...up to 8 concurrent calls

    API->>Redis: Set task stop flag
    API->>DB: Update MonitorTask status → awaiting_processing
    API-->>API: Return response to user

    LLM-->>Pool: Response for field 1 (already in flight)
    LLM-->>Pool: Response for field 2 (already in flight)
    Note over Pool: ≤8 in-flight calls complete (already paid for)

    Pool->>Redis: Field 9: MonitorTaskCacheStore.is_stopped()?
    Redis-->>Pool: Yes
    Note over Pool: Return empty ClassifyOutput (skip LLM call)
    Note over Pool: Remaining fields also skip

    Pool-->>Worker: Return all classification results

    Worker->>Worker: Filter out empty results (skipped fields)
    Worker->>DB: Write only classified fields
    Worker->>DB: Check is_monitor_task_paused() (single DB query)
    DB-->>Worker: Yes
    Worker->>DB: Reset unclassified resources (CLASSIFYING → ADDITION)
```

## Fallback: detect_interrupted_tasks

If cooperative cancellation hasn't stopped the task (e.g. all worker threads are blocked waiting on LLM responses and haven't checked the DB yet), `detect_interrupted_tasks` — which runs periodically — will detect the task is still active after a grace period (5 minutes) and log an error for admin investigation. Manual worker restart may be required.

```mermaid
flowchart TD
    A["detect_interrupted_tasks()\nruns every N seconds"] --> B["_mark_interrupted_in_progress_tasks()"]
    A --> C["_flag_unresponsive_stopped_tasks()"]

    subgraph Existing["Existing behavior"]
        B --> B1["Find tasks: status in\n{pending, in_processing}"]
        B1 --> B2{Celery worker active?}
        B2 -- Yes --> B3[OK, skip]
        B2 -- No --> B4["Mark as ERROR\ntask_error_handler()"]
    end

    subgraph New["New behavior"]
        C --> C1["Find tasks: status = awaiting_processing\nupdated 5min–24h ago"]
        C1 --> C2{Celery worker active?}
        C2 -- No --> C3["OK, cooperative cancellation\nworked — skip"]
        C2 -- Yes --> C4["Log error\nManual worker restart\nmay be required"]
    end
```
