# Sygnały aktywności

Długo działające zadania i kroki strumieniowe mogą wysyłać sygnały aktywności, aby potwierdzić, że nadal działają. Bez nich wykrywacz nieaktywnych zadań może ponownie pobrać zadanie, które nadal jest przetwarzane, ponieważ jego znacznik czasu `reserved_at` nie został zaktualizowany.

## Jak to działa

Gdy worker rozpoczyna przetwarzanie zadania, wywołuje `Heartbeat::init()`, aby ustawić kontekst bieżącego zadania. Wewnątrz handlera wywołanie `Heartbeat::beat()` aktualizuje kolumnę `reserved_at` zadania do wartości `NOW()`. Wykrywacz nieaktywnych zadań korzysta z `reserved_at`, aby ustalić, czy zadanie utknęło, więc odświeżanie tej wartości chroni zadanie.

Po zakończeniu zadania worker wywołuje `Heartbeat::clear()`, aby zresetować kontekst.

## Wysyłanie sygnałów aktywności

Wywołaj `Heartbeat::beat()` z wnętrza dowolnego handlera zadania lub kroku strumieniowego:

```php
use Queuety\Heartbeat;

class LongRunningStep implements \Queuety\Step {

    public function handle( array $state ): array {
        $items = $state['items'];

        foreach ( $items as $i => $item ) {
            $this->process( $item );
            Heartbeat::beat( [ 'processed' => $i + 1, 'total' => count( $items ) ] );
        }

        return [ 'done' => true ];
    }

    public function config(): array {
        return [];
    }

    private function process( $item ): void {
        // Expensive work here.
    }
}
```

## Dane o postępach

Parametr `$progress` jest opcjonalną tablicą asocjacyjną. Jeśli zostanie podany, jest kodowany w formacie JSON i zapisywany w kolumnie `heartbeat_data` wiersza zadania. Pozwala to śledzić postęp spoza workera:

```php
Heartbeat::beat( [
    'processed' => 450,
    'total'     => 1000,
    'percent'   => 45,
] );
```

Jeśli wywołasz tę metodę bez argumentów, aktualizowane jest tylko `reserved_at`:

```php
Heartbeat::beat();
```

## Sygnał aktywności API

Klasa `Queuety\Heartbeat` jest całkowicie statyczna. Nie twórz jej instancji.

| Metoda | Opis |
|---|---|
| `Heartbeat::init( int $job_id, Connection $conn )` | Ustawia kontekst zadania. Worker wywołuje ją automatycznie przed przetwarzaniem. |
| `Heartbeat::beat( array $progress = [] )` | Wysyła sygnał aktywności. Aktualizuje `reserved_at` i opcjonalnie zapisuje dane o postępach. |
| `Heartbeat::clear()` | Czyści kontekst zadania. Worker wywołuje ją automatycznie po przetwarzaniu. |
| `Heartbeat::current_job_id()` | Pobiera identyfikator bieżącego zadania (przydatne podczas testowania). |

## Zastosowanie w krokach strumieniowych

Worker już wysyła sygnał aktywności po każdym zapisanym fragmencie, ustawiając `heartbeat_data` na `{"streaming_chunks": n}`, więc strumień regularnie zwracający fragmenty sam utrzymuje aktualny stan. Samodzielnie wywołaj `Heartbeat::beat()`, gdy między kolejnymi fragmentami mogą wystąpić długie przerwy, na przykład podczas oczekiwania na pierwszy token. Dane przekazane przez Ciebie zostaną zastąpione przez worker przy następnym sygnale po fragmencie:

```php
use Queuety\Contracts\StreamingStep;
use Queuety\Heartbeat;

class StreamAPIResponse implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        $response = $this->client->stream( $state['prompt'] );

        foreach ( $response as $i => $chunk ) {
            Heartbeat::beat( [ 'chunks' => $i + 1 ] );
            yield $chunk;
        }
    }

    public function on_complete( array $chunks, array $state ): array {
        return [ 'response' => implode( '', $chunks ) ];
    }

    public function config(): array {
        return [];
    }
}
```

## Kiedy używać sygnałów aktywności

Używaj sygnałów aktywności, gdy pojedyncze zadanie lub krok trwa dłużej niż `QUEUETY_STALE_TIMEOUT` sekund (domyślnie: 600). Typowe przypadki:

- Wywołania LLM API z odpowiedziami strumieniowymi
- Pobieranie lub przesyłanie dużych plików
- Powtarzające się przetwarzanie wsadowe w ramach jednego kroku
- Każde zewnętrzne wywołanie API, którego odpowiedź może zająć kilka minut

Jeśli krok kończy się w mniej niż minutę, heartbeat nie jest potrzebny. Wykrywanie nieaktualnych procesów uruchamia się dopiero po upływie skonfigurowanego limitu czasu.