# Pasos de transmisión

Los pasos de transmisión permiten que un flujo de trabajo genere resultados de forma incremental. Cada fragmento generado se escribe de inmediato en la base de datos, de modo que, si el paso falla durante la transmisión, los fragmentos ya persistidos se conservan y están disponibles al reintentar.

Esto resulta útil para respuestas transmitidas por LLM, el procesamiento de archivos grandes y cualquier situación en la que no deba perderse el progreso parcial.

## La interfaz StreamingStep

Un paso de transmisión implementa `Queuety\Contracts\StreamingStep`:

```php
namespace Queuety\Contracts;

interface StreamingStep {
    public function stream( array $state, array $existing_chunks = [] ): \Generator;
    public function on_complete( array $chunks, array $state ): array;
    public function config(): array;
}
```

| Método | Descripción |
|---|---|
| `stream( $state, $existing_chunks )` | Generador que produce fragmentos de texto. Cada generación provoca una escritura inmediata en la base de datos. |
| `on_complete( $chunks, $state )` | Se llama cuando termina la transmisión. Recibe todos los fragmentos (anteriores y nuevos) y devuelve datos que se combinarán con el estado del flujo de trabajo. |
| `config()` | Configuración opcional. Admite las claves `max_attempts` y `backoff`; `backoff` acepta un nombre de estrategia (`exponential`, `linear`, `fixed`), ya que actualmente se descarta una lista de demoras ([#16](https://github.com/inline0/queuety/issues/16)). |

## Cómo se persisten los fragmentos

Cada vez que el generador produce un valor, Queuety llama a `ChunkStore::append_chunk()` para escribirlo en la tabla `queuety_chunks`. No se agrupan ni almacenan en búfer. Cada fragmento generado equivale a una operación INSERT.

La fila del fragmento almacena `job_id`, `workflow_id`, `step_index`, `chunk_index` (desde cero) y la cadena `content`.

## Reanudación

Cuando se reintenta un paso de transmisión tras un fallo, el worker carga todos los fragmentos persistidos y los pasa como parámetro `$existing_chunks`. Tu implementación determina cómo reanudar:

- **Omitir los elementos ya procesados** comprobando el número de fragmentos existentes
- **Avanzar en un API paginado** usando el número de fragmentos como desplazamiento de página
- **Empezar desde cero** ignorando por completo `$existing_chunks` (los fragmentos del intento anterior siguen en la base de datos y se añadirán al principio)

```php
public function stream( array $state, array $existing_chunks = [] ): \Generator {
    $offset = count( $existing_chunks );

    $pages = $this->client->paginate( $state['query'], offset: $offset );

    foreach ( $pages as $page ) {
        yield json_encode( $page );
    }
}
```

## Señales de actividad durante la transmisión

Las transmisiones largas deben enviar señales de actividad para evitar que el detector de trabajos obsoletos recupere el trabajo. Llama a `Heartbeat::beat()` dentro del bucle del generador:

```php
use Queuety\Heartbeat;

public function stream( array $state, array $existing_chunks = [] ): \Generator {
    foreach ( $this->client->stream( $state['prompt'] ) as $i => $chunk ) {
        Heartbeat::beat( [ 'chunks' => $i + 1 ] );
        yield $chunk;
    }
}
```

Consulta [Señales de actividad](https://queuety.inline0.dev/es/docs/features/heartbeats) para obtener más información sobre la API de actividad.

## Limpieza de fragmentos

Cuando la transmisión termina correctamente y `on_complete()` devuelve un resultado, el worker llama a `ChunkStore::clear_chunks()` para eliminar todos los fragmentos del trabajo. El resultado combinado de `on_complete()` es el registro duradero. Los fragmentos son temporales por diseño.

Si el paso falla y se reintenta, los fragmentos no se eliminan. Se vuelven a pasar mediante `$existing_chunks` en el siguiente intento.

## Ejemplo completo: transmisión de LLM

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

class StreamLLMResponse implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        // If we have existing chunks from a previous attempt, we can skip
        // re-requesting. The on_complete method will concatenate everything.
        if ( ! empty( $existing_chunks ) ) {
            return;
        }

        $response = OpenAI::chat()->createStreamed( [
            'model'    => 'gpt-4',
            'messages' => $state['messages'],
        ] );

        foreach ( $response as $i => $chunk ) {
            $text = $chunk->choices[0]->delta->content ?? '';
            if ( '' !== $text ) {
                Heartbeat::beat( [ 'tokens' => $i ] );
                yield $text;
            }
        }
    }

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

    public function config(): array {
        return [ 'max_attempts' => 5 ];
    }
}
```

Úsalo en un flujo de trabajo como cualquier otro paso:

```php
use Queuety\Queuety;

Queuety::workflow( 'ai_chat' )
    ->then( BuildPromptHandler::class )
    ->then( StreamLLMResponse::class )
    ->then( SaveResponseHandler::class )
    ->dispatch( [ 'user_input' => $input ] );
```

## Ejemplo completo: procesamiento de archivos

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

class ProcessCSVRows implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        $file   = fopen( $state['csv_path'], 'r' );
        $offset = count( $existing_chunks );
        $row    = 0;

        // Skip the header.
        fgetcsv( $file );

        while ( $line = fgetcsv( $file ) ) {
            $row++;

            // Skip rows already processed in a previous attempt.
            if ( $row <= $offset ) {
                continue;
            }

            $result = $this->process_row( $line );
            Heartbeat::beat( [ 'row' => $row ] );
            yield json_encode( $result );
        }

        fclose( $file );
    }

    public function on_complete( array $chunks, array $state ): array {
        $results = array_map( 'json_decode', $chunks );
        return [
            'processed_count' => count( $results ),
            'results'         => $results,
        ];
    }

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

    private function process_row( array $line ): array {
        // Your row processing logic here.
        return [ 'id' => $line[0], 'status' => 'ok' ];
    }
}
```

## Clase ChunkStore API

La clase `ChunkStore` gestiona la persistencia de fragmentos. Accede a ella mediante `Queuety::chunk_store()`.

```php
$store = Queuety::chunk_store();
```

| Método | Devuelve | Descripción |
|---|---|---|
| `get_chunks( int $job_id )` | `string[]` | Obtiene todos los fragmentos de un trabajo, ordenados por índice de fragmento |
| `append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )` | `void` | Añade un único fragmento |
| `clear_chunks( int $job_id )` | `void` | Elimina todos los fragmentos de un trabajo |
| `chunk_count( int $job_id )` | `int` | Cuenta los fragmentos de un trabajo |
| `get_accumulated( int $job_id )` | `string` | Concatena todos los fragmentos en una sola cadena |

Normalmente no necesitas llamar directamente a `ChunkStore`. El worker gestiona automáticamente la persistencia y la limpieza al procesar un `StreamingStep`. API está disponible para casos avanzados, como inspeccionar fragmentos en curso o realizar una limpieza manual.