diff --git a/README.md b/README.md
index 15d4e37..31086ea 100644
--- a/README.md
+++ b/README.md
@@ -118,6 +118,20 @@ $table->addDocuments([
     ]); 
 ```
 
+### Fast bulk insert (indexer-assisted):
+
+Requires a Manticore build that supports `/bulk?indexer_rt_bulk=1`. Documents must include explicit non-zero numeric ids.
+
+```php
+$result = $table->addDocumentsStreaming([
+        ['id'=>2,'title'=>'Interstellar','plot'=>'A team of explorers travel through a wormhole in space in an attempt to ensure humanity\'s survival.','_year'=>2014,'rating'=>8.5],
+        ['id'=>3,'title'=>'Inception','plot'=>'A thief who steals corporate secrets through the use of dream-sharing technology is given the inverse task of planting an idea into the mind of a C.E.O.','_year'=>2010,'rating'=>8.8],
+    ], [
+        // omit workers to auto-tune 1,2,4,... up to server workers_total
+        'max_workers' => 8,
+    ]);
+```
+
 ### Perform a search:
 
 ```php
diff --git a/docs/configuration.md b/docs/configuration.md
index 555ca62..790eab7 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -48,6 +48,14 @@ Connection and transport options:
 * timeout - query timeout
 * connection_timeout - connection timeout
 * proxy - proxy definition as host:port
+
+`Client::indexerBulk()` / `Table::addDocumentsStreaming()` require the curl-based `Http` or `Https` transport. They open additional curl handles for concurrent chunked uploads that share one document iterator and drain it until empty (no batching on the main path).
+
+For large assisted bulk loads:
+
+* Prefer raising the connection `timeout` (default 300s) if indexer builds are slow
+* Server `max_packet_size` is bypassed by chunked `/bulk` uploads, but reverse proxies may still impose their own body limits
+* Cap client concurrency with `max_workers`; the client also reads `workers_total` from `SHOW STATUS`
 * username - username for HTTP Basic authentication
 * password - password for HTTP Basic authentication
 * bearer_token - bearer token used in the `Authorization` header
diff --git a/docs/lowlevelclient.md b/docs/lowlevelclient.md
index b4c5ebd..f665a07 100644
--- a/docs/lowlevelclient.md
+++ b/docs/lowlevelclient.md
@@ -402,4 +402,28 @@ $doc = [
 $response = $client->bulk($doc);
 ```
 
-<!-- proofread -->
\ No newline at end of file
+### Indexer-assisted bulk (`indexerBulk`)
+
+Experimental high-speed append path that posts to `/bulk?indexer_rt_bulk=1`. Prefer the Table helper `addDocumentsStreaming()` for document lists; use `indexerBulk()` for raw insert operations.
+
+```php
+$result = $client->indexerBulk([
+    ['insert' => ['table' => 'products', 'id' => 101, 'doc' => ['title' => 'Bag', 'price' => 9.99]]],
+    ['insert' => ['table' => 'products', 'id' => 102, 'doc' => ['title' => 'Baz', 'price' => 4.5]]],
+], [
+    'workers' => 2,
+]);
+```
+
+Notes:
+
+- Requires curl `Http`/`Https` transport. `PhpHttp` is not supported for this path.
+- IDs must be non-zero PHP integers or decimal digit strings in the uint64 range. Pass IDs above `PHP_INT_MAX` as strings, for example `'18446744073709551615'`; floats are rejected.
+- Opens up to `workers` concurrent chunked streams that share one iterator and pull until every document is inserted (no batching on the main path).
+- `probe_docs` applies only to auto-tune probe uploads when `workers` is omitted (default 1000 docs per worker).
+- PHP `post_max_size` / `upload_max_filesize` do not apply to outgoing client requests; `memory_limit`, proxies, and timeouts still do.
+- When `workers` is omitted, concurrency is probed as 1, 2, 4, … up to `workers_total` from `SHOW STATUS` (or `max_workers`).
+- Do not retry an ambiguous network failure after a request may have been accepted: each assisted request is atomic, but a lost response does not tell you whether the disk chunk was attached.
+- Server support depends on a Manticore build that includes indexer-assisted RT bulk; see upstream documentation for the released version that ships the feature.
+
+<!-- proofread -->
diff --git a/docs/tableclass.md b/docs/tableclass.md
index 8557453..70825dc 100755
--- a/docs/tableclass.md
+++ b/docs/tableclass.md
@@ -137,6 +137,46 @@ Returns an array response with:
 - errors - stating whether an error occurred
 - items - response status for each document.
 
+### addDocumentsStreaming()
+
+High-throughput append-only insert using Manticore's indexer-assisted bulk mode (`/bulk?indexer_rt_bulk=1`).
+
+Each request is streamed as NDJSON with HTTP chunked transfer encoding (documents are encoded on demand into a small buffer).
+
+With `workers = N`, the client opens up to N concurrent streams that share one document iterator and pull until every document is inserted. Each finished stream commits as one disk chunk. 
+
+Requirements and limits (server-side):
+
+- Local real-time table only (not distributed, sharded, replicated, percolate, or plain)
+- Insert only; replace/update/delete are rejected
+- Clustered tables are not supported
+- Every document must have an explicit non-zero numeric `id`
+
+IDs can be PHP integers or decimal digit strings up to `18446744073709551615`. Pass IDs above `PHP_INT_MAX` as strings to preserve their exact value; floats are rejected. Leading zeros are removed before duplicate detection.
+
+Options:
+
+- `workers` - fixed concurrency; when omitted, the client probes `1,2,4,...` until throughput stops improving
+- `max_workers` - upper bound (also capped by Manticore's `workers_total` setting)
+- `probe_docs` - documents per worker during each auto-tune probe (default 1000)
+- `gain_threshold` - minimum relative throughput gain to keep doubling workers (default `0`; stop when the next step is not better)
+- `timeout` - per-request timeout override
+- `fallback` - if true, fall back to regular `/bulk` when assisted mode is unavailable
+
+Example:
+
+```php
+$result = $table->addDocumentsStreaming([
+    ['id' => 1, 'title' => 'Crossbody Bag', 'gid' => 10],
+    ['id' => 2, 'title' => 'Sheet Set', 'gid' => 11],
+], [
+    // 'workers' => 4, // set explicitly to skip auto-tuning
+    'max_workers' => 8,
+]);
+
+// Persist $result['selected_workers'] for later runs if auto-tuning was used.
+```
+
 ### getDocumentById()
 
 Get an existing document by its ID.
@@ -698,4 +738,4 @@ Array
         )
 )
 ```
-<!-- proofread -->
\ No newline at end of file
+<!-- proofread -->
diff --git a/src/Manticoresearch/Bulk/AdaptiveTuner.php b/src/Manticoresearch/Bulk/AdaptiveTuner.php
new file mode 100644
index 0000000..8d6d2ab
--- /dev/null
+++ b/src/Manticoresearch/Bulk/AdaptiveTuner.php
@@ -0,0 +1,123 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Bulk;
+
+/**
+ * Chooses concurrent request count by exponential probing of throughput.
+ */
+class AdaptiveTuner
+{
+	/** @var float */
+	private $gainThreshold;
+
+	/** @var int */
+	private $maxWorkers;
+
+	/**
+	 * @param float $gainThreshold
+	 * @param int $maxWorkers
+	 */
+	public function __construct(float $gainThreshold, int $maxWorkers) {
+		$this->gainThreshold = $gainThreshold;
+		$this->maxWorkers = max(1, $maxWorkers);
+	}
+
+	/**
+	 * Yield candidate worker counts: 1, 2, 4, ... up to max.
+	 *
+	 * @return int[]
+	 */
+	public function candidates(): array {
+		$values = [];
+		for ($n = 1; $n <= $this->maxWorkers; $n *= 2) {
+			$values[] = $n;
+			if ($n > intdiv($this->maxWorkers, 2)) {
+				break;
+			}
+		}
+		if ($values[sizeof($values) - 1] !== $this->maxWorkers) {
+			$values[] = $this->maxWorkers;
+		}
+		return array_values(array_unique($values));
+	}
+
+	/**
+	 * Decide whether a newer probe is better enough to keep searching.
+	 *
+	 * @param float $previousThroughput docs/sec (or bytes/sec)
+	 * @param float $currentThroughput
+	 * @return bool true when current is an improvement worth continuing from
+	 */
+	public function isImprovement(float $previousThroughput, float $currentThroughput): bool {
+		if ($previousThroughput <= 0) {
+			return $currentThroughput > 0;
+		}
+		return ($currentThroughput - $previousThroughput) / $previousThroughput >= $this->gainThreshold;
+	}
+
+	/**
+	 * Pick the worker count with the best throughput; ties prefer fewer workers.
+	 *
+	 * @param array<int,float> $throughputByWorkers worker => throughput
+	 * @return int
+	 */
+	public function selectBest(array $throughputByWorkers): int {
+		if ($throughputByWorkers === []) {
+			return 1;
+		}
+		$bestWorkers = 1;
+		$bestRate = -1.0;
+		ksort($throughputByWorkers);
+		foreach ($throughputByWorkers as $workers => $rate) {
+			if ($rate <= $bestRate) {
+				continue;
+			}
+
+			$bestRate = $rate;
+			$bestWorkers = (int)$workers;
+		}
+		return max(1, min($bestWorkers, $this->maxWorkers));
+	}
+
+	/**
+	 * Walk candidates using a callable probe: fn(int $workers): float throughput.
+	 * Stops early when improvement falls below the gain threshold.
+	 *
+	 * @param callable $probe
+	 * @return array{selected_workers:int,stages:array}
+	 */
+	public function tune(callable $probe): array {
+		$stages = [];
+		$rates = [];
+		$previous = null;
+
+		foreach ($this->candidates() as $workers) {
+			$rate = (float)$probe($workers);
+			$rates[$workers] = $rate;
+			$stages[] = [
+				'workers' => $workers,
+				'throughput' => $rate,
+			];
+			if ($previous === null) {
+				$previous = $rate;
+				continue;
+			}
+			if ($this->isImprovement($previous, $rate)) {
+				$previous = $rate;
+				continue;
+			}
+			// No meaningful gain — stop probing higher concurrency
+			break;
+		}
+
+		return [
+			'selected_workers' => $this->selectBest($rates),
+			'stages' => $stages,
+		];
+	}
+}
diff --git a/src/Manticoresearch/Bulk/CurlUploader.php b/src/Manticoresearch/Bulk/CurlUploader.php
new file mode 100644
index 0000000..0049b51
--- /dev/null
+++ b/src/Manticoresearch/Bulk/CurlUploader.php
@@ -0,0 +1,313 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Bulk;
+
+use Manticoresearch\Connection;
+use Manticoresearch\Exceptions\ConnectionException;
+use Manticoresearch\Exceptions\ResponseException;
+use Manticoresearch\Exceptions\RuntimeException;
+use Manticoresearch\Request;
+use Manticoresearch\Response\Bulk as BulkResponse;
+use Manticoresearch\Transport;
+
+/**
+ * Concurrent chunked HTTP uploads to /bulk?indexer_rt_bulk=1 via curl_multi.
+ *
+ * Each request body is produced incrementally by NdjsonStream::read() — the full
+ * NDJSON payload is never concatenated in memory.
+ */
+class CurlUploader
+{
+	/** @var Connection */
+	private $connection;
+
+	/** @var int|null */
+	private $timeout;
+
+	/**
+	 * @param Connection $connection
+	 * @param int|null $timeout
+	 */
+	public function __construct(Connection $connection, ?int $timeout = null) {
+		$this->connection = $connection;
+		$this->timeout = $timeout;
+	}
+
+	/**
+	 * @return void
+	 */
+	public function assertSupportedTransport(): void {
+		$transport = $this->connection->getTransport();
+		// phpcs:ignore SlevomatCodingStandard.Classes.ModernClassNameReference.ClassNameReferencedViaFunctionCall
+		$name = is_object($transport) ? get_class($transport) : (string)$transport;
+		$base = $name;
+		if (strpos($name, '\\') !== false) {
+			$parts = explode('\\', $name);
+			$base = end($parts);
+		}
+		if (!in_array($base, ['Http', 'Https'], true)) {
+			throw new RuntimeException(
+				'indexerBulk requires Http/Https (curl) transport; got ' . $base
+			);
+		}
+	}
+
+	/**
+	 * Upload NDJSON streams concurrently (up to $workers in flight).
+	 *
+	 * When $phases is provided, accumulates:
+	 * - http: curl_multi wait time excluding read-callback work
+	 * - decode: response parse time in finalizeHandle
+	 *
+	 * @param NdjsonStream[] $streams
+	 * @param int $workers
+	 * @param array<string,float>|null $phases
+	 * @return array{responses:array,docs:int,bytes:int,requests:int,elapsed:float}
+	 */
+	public function uploadStreams(array $streams, int $workers, &$phases = null): array {
+		$this->assertSupportedTransport();
+		$workers = max(1, $workers);
+		$queue = [];
+		foreach ($streams as $stream) {
+			if (!($stream instanceof NdjsonStream) || $stream->isEmpty()) {
+				continue;
+			}
+
+			$queue[] = $stream;
+		}
+		if ($queue === []) {
+			return [
+				'responses' => [],
+				'docs' => 0,
+				'bytes' => 0,
+				'requests' => 0,
+				'elapsed' => 0.0,
+			];
+		}
+
+		$start = microtime(true);
+		$multi = curl_multi_init();
+		if ($multi === false) {
+			throw new RuntimeException('Failed to initialize curl_multi');
+		}
+
+		$active = [];
+		$responses = [];
+		$totalDocs = 0;
+		$totalBytes = 0;
+		$requestCount = 0;
+		$nextId = 0;
+		$trackPhases = is_array($phases);
+		$readCallbackTime = 0.0;
+
+		try {
+			while ($queue !== [] || $active !== []) {
+				while (sizeof($active) < $workers && $queue !== []) {
+					$stream = array_shift($queue);
+					$handleId = $nextId++;
+					$state = $this->createHandle($stream, $handleId, $readCallbackTime);
+					$active[$handleId] = $state;
+					curl_multi_add_handle($multi, $state['ch']);
+				}
+
+				$readBefore = $readCallbackTime;
+				$httpStart = microtime(true);
+				do {
+					$status = curl_multi_exec($multi, $running);
+				} while ($status === CURLM_CALL_MULTI_PERFORM);
+
+				if ($status !== CURLM_OK) {
+					throw new RuntimeException('curl_multi_exec failed with status ' . $status);
+				}
+				if ($trackPhases) {
+					$phases['http'] += microtime(true) - $httpStart - ($readCallbackTime - $readBefore);
+				}
+
+				while ($info = curl_multi_info_read($multi)) {
+					$ch = $info['handle'];
+					$handleId = $this->findHandleId($active, $ch);
+					if ($handleId === null) {
+						curl_multi_remove_handle($multi, $ch);
+						curl_close($ch);
+						continue;
+					}
+					$state = $active[$handleId];
+					$decodeStart = $trackPhases ? microtime(true) : 0.0;
+					$response = $this->finalizeHandle($state, $info['result']);
+					if ($trackPhases) {
+						$phases['decode'] += microtime(true) - $decodeStart;
+					}
+					$responses[] = $response;
+					$totalDocs += $response['docs'];
+					$totalBytes += $response['bytes'];
+					$requestCount++;
+					curl_multi_remove_handle($multi, $ch);
+					curl_close($ch);
+					unset($active[$handleId]);
+				}
+
+				if (!$running || $active === []) {
+					continue;
+				}
+
+				$httpStart = microtime(true);
+				curl_multi_select($multi, 1.0);
+				if (!$trackPhases) {
+					continue;
+				}
+				$phases['http'] += microtime(true) - $httpStart;
+			}
+		} finally {
+			foreach ($active as $state) {
+				curl_multi_remove_handle($multi, $state['ch']);
+				curl_close($state['ch']);
+			}
+			curl_multi_close($multi);
+		}
+
+		return [
+			'responses' => $responses,
+			'docs' => $totalDocs,
+			'bytes' => $totalBytes,
+			'requests' => $requestCount,
+			'elapsed' => microtime(true) - $start,
+		];
+	}
+
+	/**
+	 * @param NdjsonStream $stream
+	 * @param int $handleId
+	 * @param float $readCallbackTime
+	 * @return array
+	 */
+	private function createHandle(NdjsonStream $stream, int $handleId, float &$readCallbackTime): array {
+		$connection = $this->connection;
+		$scheme = $connection->getConfig('scheme') ?: 'http';
+		$transport = $connection->getTransport();
+		if ($transport === 'Https' || $scheme === 'https') {
+			$scheme = 'https';
+		}
+		$url = $scheme . '://' . $connection->getHost() . ':' . $connection->getPort()
+			. $connection->getPath() . '/bulk?indexer_rt_bulk=1';
+
+		$ch = curl_init();
+		$request = new Request(
+			[
+				'body' => '[indexer_rt_bulk stream]',
+				'query' => ['indexer_rt_bulk' => 1],
+				'content_type' => 'application/x-ndjson',
+			]
+		);
+		$headers = (new Transport($connection))->getRequestHeadersAsList($request, $connection);
+		$headers = array_values(
+			array_filter(
+				$headers,
+				static function ($header) {
+					return stripos((string)$header, 'Content-Length:') !== 0
+						&& stripos((string)$header, 'Transfer-Encoding:') !== 0;
+				}
+			)
+		);
+		$headers[] = 'Transfer-Encoding: chunked';
+		$headers[] = 'Expect:';
+
+		$timeout = $this->timeout ?? $connection->getTimeout();
+		curl_setopt($ch, CURLOPT_URL, $url);
+		curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
+		curl_setopt($ch, CURLOPT_CUSTOMREQUEST, 'POST');
+		curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
+		curl_setopt($ch, CURLOPT_TIMEOUT, $timeout);
+		curl_setopt($ch, CURLOPT_UPLOAD, true);
+		// Omit CURLOPT_INFILESIZE so libcurl sends Transfer-Encoding: chunked.
+		curl_setopt(
+			$ch,
+			CURLOPT_READFUNCTION,
+			// phpcs:ignore Generic.CodeAnalysis.UnusedFunctionParameter,SlevomatCodingStandard.Functions.UnusedParameter
+			static function ($ch, $fd, $size) use ($stream, &$readCallbackTime) {
+				$t = microtime(true);
+				$data = $stream->read($size);
+				$readCallbackTime += microtime(true) - $t;
+				return $data;
+			}
+		);
+
+		if ($connection->getConnectTimeout() > 0) {
+			curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, $connection->getConnectTimeout());
+		}
+		if ($connection->getConfig('proxy') !== null) {
+			curl_setopt($ch, CURLOPT_PROXY, $connection->getConfig('proxy'));
+		}
+		if (!empty($connection->getConfig('curl'))) {
+			foreach ($connection->getConfig('curl') as $k => $v) {
+				curl_setopt($ch, $k, $v);
+			}
+		}
+
+		return [
+			'id' => $handleId,
+			'ch' => $ch,
+			'url' => $url,
+			'headers' => $headers,
+			'stream' => $stream,
+			'request' => $request,
+		];
+	}
+
+	/**
+	 * @param array $active
+	 * @param resource|\CurlHandle $ch
+	 * @return int|null
+	 */
+	private function findHandleId(array $active, $ch) {
+		foreach ($active as $id => $state) {
+			if ($state['ch'] === $ch) {
+				return $id;
+			}
+		}
+		return null;
+	}
+
+	/**
+	 * @param array $state
+	 * @param int $curlResult
+	 * @return array
+	 */
+	private function finalizeHandle(array $state, int $curlResult): array {
+		$ch = $state['ch'];
+		/** @var NdjsonStream $stream */
+		$stream = $state['stream'];
+		$responseString = curl_multi_getcontent($ch);
+		$status = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
+		$errorno = curl_errno($ch);
+		if ($curlResult !== CURLE_OK || $errorno > 0) {
+			$error = curl_error($ch) ?: ('curl error ' . ($curlResult ?: $errorno));
+			// Ambiguous after bytes may have left the client: do not auto-retry.
+			throw new ConnectionException($error, $state['request']);
+		}
+
+		$response = new BulkResponse($responseString, $status);
+		$response->setTransportInfo(
+			[
+				'url' => $state['url'],
+				'headers' => Connection::redactAuthHeaders($state['headers']),
+				'body' => '[indexer_rt_bulk ' . $stream->getDocs() . ' docs streamed]',
+			]
+		);
+		if ($status < 200 || $status >= 300 || $response->hasError()) {
+			throw new ResponseException($state['request'], $response);
+		}
+
+		return [
+			'status' => $status,
+			'body' => $response->getResponse(),
+			'docs' => $stream->getDocs(),
+			'bytes' => $stream->getBytes(),
+			'table' => $stream->getTable(),
+		];
+	}
+}
diff --git a/src/Manticoresearch/Bulk/IndexerBulk.php b/src/Manticoresearch/Bulk/IndexerBulk.php
new file mode 100644
index 0000000..2126664
--- /dev/null
+++ b/src/Manticoresearch/Bulk/IndexerBulk.php
@@ -0,0 +1,344 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Bulk;
+
+use Manticoresearch\Client;
+use Manticoresearch\Exceptions\ResponseException;
+use Manticoresearch\Exceptions\RuntimeException;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+
+/**
+ * Orchestrates indexer-assisted bulk ingestion with optional adaptive concurrency.
+ */
+class IndexerBulk
+{
+	/** @var Client */
+	private $client;
+
+	/** @var LoggerInterface */
+	private $logger;
+
+	/** @var OperationNormalizer|null */
+	private $normalizer;
+
+	/**
+	 * @param Client $client
+	 * @param LoggerInterface|null $logger
+	 * @param OperationNormalizer|null $normalizer
+	 */
+	public function __construct(
+		Client $client,
+		?LoggerInterface $logger = null,
+		?OperationNormalizer $normalizer = null
+	) {
+		$this->client = $client;
+		$this->logger = $logger ?? new NullLogger();
+		$this->normalizer = $normalizer;
+	}
+
+	/**
+	 * @param iterable $operations
+	 * @param array $options
+	 * @return array
+	 */
+	public function run(iterable $operations, array $options = []): array {
+		$phases = null;
+		if (array_key_exists('phases', $options)) {
+			$phases = &$options['phases'];
+			unset($options['phases']);
+		}
+		$opts = Options::fromArray($options);
+		if ($opts->fallback && !is_array($operations)) {
+			$operations = iterator_to_array($operations, false);
+		}
+		$normalizer = $this->normalizer ?? new OperationNormalizer();
+		$source = new OperationIterator($operations, $normalizer);
+
+		try {
+			$connection = $this->client->getConnectionPool()->getConnection();
+			$uploader = new CurlUploader($connection, $opts->timeout);
+			$uploader->assertSupportedTransport();
+		} catch (RuntimeException $e) {
+			if ($opts->fallback) {
+				return $this->fallbackBulk($operations, $e->getMessage());
+			}
+			throw $e;
+		}
+
+		$warnings = [];
+		$cap = new WorkerCap($this->client);
+		$maxWorkers = $cap->resolveMax($opts->maxWorkers);
+
+		$uploaded = [
+			'responses' => [],
+			'docs' => 0,
+			'bytes' => 0,
+			'requests' => 0,
+			'elapsed' => 0.0,
+		];
+		$tuning = [
+			'stages' => [],
+		];
+
+		if ($opts->workers !== null) {
+			$selectedWorkers = max(1, min($opts->workers, $maxWorkers));
+			if ($opts->workers > $maxWorkers) {
+				$warnings[] = 'Requested workers=' . $opts->workers
+					. ' capped to max_workers=' . $maxWorkers;
+			}
+		} else {
+			try {
+				$tuner = new AdaptiveTuner($opts->gainThreshold, $maxWorkers);
+				$tuned = $this->autoTune($source, $uploader, $tuner, $opts, $warnings, $phases);
+			} catch (ResponseException $e) {
+				if ($opts->fallback && $this->isCapabilityError($e)) {
+					return $this->fallbackBulk($operations, $e->getMessage());
+				}
+				throw $e;
+			} catch (RuntimeException $e) {
+				if ($opts->fallback && $this->isCapabilityErrorMessage($e->getMessage())) {
+					return $this->fallbackBulk($operations, $e->getMessage());
+				}
+				throw $e;
+			}
+			$selectedWorkers = $tuned['selected_workers'];
+			$tuning = [
+				'stages' => $tuned['stages'],
+			];
+			$uploaded['responses'] = $tuned['probe_responses'];
+			$uploaded['docs'] = $tuned['probe_docs'];
+			$uploaded['bytes'] = $tuned['probe_bytes'];
+			$uploaded['requests'] = $tuned['probe_requests'];
+			$uploaded['elapsed'] = $tuned['probe_elapsed'];
+		}
+
+		try {
+			$rest = $this->uploadAll($source, $uploader, $selectedWorkers, $phases);
+		} catch (ResponseException $e) {
+			if ($opts->fallback && $uploaded['docs'] === 0 && $this->isCapabilityError($e)) {
+				return $this->fallbackBulk($operations, $e->getMessage());
+			}
+			throw $e;
+		}
+
+		$uploaded['responses'] = array_merge($uploaded['responses'], $rest['responses']);
+		$uploaded['docs'] += $rest['docs'];
+		$uploaded['bytes'] += $rest['bytes'];
+		$uploaded['requests'] += $rest['requests'];
+		$uploaded['elapsed'] += $rest['elapsed'];
+
+		$throughput = ((float)$uploaded['elapsed']) > 0.0
+			? $uploaded['docs'] / $uploaded['elapsed']
+			: 0.0;
+
+		return [
+			'items' => $uploaded['responses'],
+			'docs' => $uploaded['docs'],
+			'bytes' => $uploaded['bytes'],
+			'requests' => $uploaded['requests'],
+			'elapsed' => $uploaded['elapsed'],
+			'throughput' => $throughput,
+			'selected_workers' => $selectedWorkers,
+			'max_workers' => $maxWorkers,
+			'tuning' => $tuning,
+			'warnings' => $warnings,
+			'mode' => 'indexer_rt_bulk',
+		];
+	}
+
+	/**
+	 * @param OperationIterator $source
+	 * @param CurlUploader $uploader
+	 * @param AdaptiveTuner $tuner
+	 * @param Options $opts
+	 * @param array $warnings
+	 * @return array
+	 */
+	private function autoTune(
+		OperationIterator $source,
+		CurlUploader $uploader,
+		AdaptiveTuner $tuner,
+		Options $opts,
+		array &$warnings,
+		&$phases = null
+	): array {
+		$probeResponses = [];
+		$probeDocs = 0;
+		$probeBytes = 0;
+		$probeRequests = 0;
+		$probeElapsed = 0.0;
+
+		$probe = function (int $workers) use (
+			$source,
+			$uploader,
+			$opts,
+			&$probeResponses,
+			&$probeDocs,
+			&$probeBytes,
+			&$probeRequests,
+			&$probeElapsed,
+			&$warnings,
+			&$phases
+		) {
+			$streams = $this->takeStreams($source, $workers, $opts->probeDocs);
+			if ($streams === []) {
+				return 0.0;
+			}
+			try {
+				$result = $uploader->uploadStreams($streams, sizeof($streams), $phases);
+			} catch (ResponseException $e) {
+				if ($this->isCapabilityError($e)) {
+					$warnings[] = $e->getMessage();
+				}
+				throw $e;
+			}
+			$probeResponses = array_merge($probeResponses, $result['responses']);
+			$probeDocs += $result['docs'];
+			$probeBytes += $result['bytes'];
+			$probeRequests += $result['requests'];
+			$probeElapsed += $result['elapsed'];
+			return $result['elapsed'] > 0 ? ($result['docs'] / $result['elapsed']) : 0.0;
+		};
+
+		$tuning = $tuner->tune($probe);
+		$tuning['probe_responses'] = $probeResponses;
+		$tuning['probe_docs'] = $probeDocs;
+		$tuning['probe_bytes'] = $probeBytes;
+		$tuning['probe_requests'] = $probeRequests;
+		$tuning['probe_elapsed'] = $probeElapsed;
+		return $tuning;
+	}
+
+	/**
+	 * Open `workers` streams that share the source and pull until it is empty.
+	 *
+	 * @param OperationIterator $source
+	 * @param CurlUploader $uploader
+	 * @param int $workers
+	 * @return array
+	 */
+	private function uploadAll(
+		OperationIterator $source,
+		CurlUploader $uploader,
+		int $workers,
+		&$phases = null
+	): array {
+		// maxDocs = 0 → each worker drains until the shared iterator is empty.
+		$streams = $this->takeStreams($source, $workers, 0);
+		if ($streams === []) {
+			return [
+				'responses' => [],
+				'docs' => 0,
+				'bytes' => 0,
+				'requests' => 0,
+				'elapsed' => 0.0,
+			];
+		}
+		return $uploader->uploadStreams($streams, sizeof($streams), $phases);
+	}
+
+	/**
+	 * Open up to $count streams that share $source (pulling happens during upload).
+	 *
+	 * @param OperationIterator $source
+	 * @param int $count
+	 * @param int $maxDocs 0 = unlimited
+	 * @return NdjsonStream[]
+	 */
+	private function takeStreams(
+		OperationIterator $source,
+		int $count,
+		int $maxDocs
+	): array {
+		$streams = [];
+		for ($i = 0; $i < $count; $i++) {
+			$stream = NdjsonStream::tryCreate($source, $maxDocs);
+			if ($stream === null) {
+				break;
+			}
+			$streams[] = $stream;
+		}
+		return $streams;
+	}
+
+	/**
+	 * @param iterable $operations
+	 * @param string $reason
+	 * @return array
+	 */
+	private function fallbackBulk(iterable $operations, string $reason): array {
+		$this->logger->warning('Falling back to regular bulk: ' . $reason);
+		// Keep pre-encoded uint64 IDs numeric when passing through the regular bulk endpoint.
+		$body = '';
+		$docs = 0;
+		$normalizer = $this->normalizer ?? new OperationNormalizer(null, false);
+		foreach ($operations as $operation) {
+			try {
+				$payload = $normalizer->normalize($operation);
+			} catch (RuntimeException $e) {
+				if (is_array($operation) && isset($operation['insert'])) {
+					$body .= json_encode($operation, JSON_HEX_TAG | JSON_THROW_ON_ERROR) . "\n";
+					$docs++;
+					continue;
+				}
+				throw $e;
+			}
+			$body .= $normalizer->encodeLine($payload) . "\n";
+			$docs++;
+		}
+		$response = $this->client->bulk(['body' => $body]);
+		return [
+			'items' => isset($response['items']) ? $response['items'] : $response,
+			'docs' => $docs,
+			'bytes' => null,
+			'requests' => 1,
+			'elapsed' => null,
+			'throughput' => null,
+			'selected_workers' => 1,
+			'max_workers' => 1,
+			'tuning' => ['stages' => []],
+			'warnings' => ['Fell back to regular /bulk: ' . $reason],
+			'mode' => 'bulk_fallback',
+		];
+	}
+
+	/**
+	 * @param ResponseException $e
+	 * @return bool
+	 */
+	private function isCapabilityError(ResponseException $e): bool {
+		return $this->isCapabilityErrorMessage((string)$e->getMessage());
+	}
+
+	/**
+	 * @param string $message
+	 * @return bool
+	 */
+	private function isCapabilityErrorMessage(string $message): bool {
+		$message = strtolower($message);
+		$needles = [
+			'indexer_rt_bulk',
+			'indexer rt bulk',
+			'unknown option',
+			'unknown query',
+			'unsupported',
+			'unavailable',
+			'failed to start indexer',
+			'failed to locate',
+			'sibling',
+			'requires http/https',
+		];
+		foreach ($needles as $needle) {
+			if (strpos($message, $needle) !== false) {
+				return true;
+			}
+		}
+		return false;
+	}
+}
diff --git a/src/Manticoresearch/Bulk/NdjsonStream.php b/src/Manticoresearch/Bulk/NdjsonStream.php
new file mode 100644
index 0000000..ff5244f
--- /dev/null
+++ b/src/Manticoresearch/Bulk/NdjsonStream.php
@@ -0,0 +1,169 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Bulk;
+
+use Manticoresearch\Exceptions\RuntimeException;
+
+/**
+ * Incremental NDJSON body for one /bulk?indexer_rt_bulk=1 request.
+ *
+ * Encodes documents on demand into a small leftover buffer — never builds the
+ * full request body as a single string.
+ *
+ * When $maxDocs is 0, the stream keeps pulling from the shared source until EOF
+ * (or a table change).
+ */
+class NdjsonStream
+{
+	/** @var OperationIterator */
+	private $source;
+
+	/** @var int 0 = unlimited */
+	private $maxDocs;
+
+	/** @var string */
+	private $buffer = '';
+
+	/** @var int */
+	private $docs = 0;
+
+	/** @var int */
+	private $bytes = 0;
+
+	/** @var array<int|string,bool> */
+	private $ids = [];
+
+	/** @var string|null */
+	private $table;
+
+	/** @var bool */
+	private $closed = false;
+
+	/** @var bool */
+	private $empty = true;
+
+	/**
+	 * @param OperationIterator $source
+	 * @param int $maxDocs 0 = pull until source EOF
+	 */
+	public function __construct(OperationIterator $source, int $maxDocs = 0) {
+		$this->source = $source;
+		$this->maxDocs = max(0, $maxDocs);
+		$this->prime();
+	}
+
+	/**
+	 * Create a stream only when at least one document is available.
+	 *
+	 * @param OperationIterator $source
+	 * @param int $maxDocs
+	 * @return self|null
+	 */
+	public static function tryCreate(OperationIterator $source, int $maxDocs = 0) {
+		$stream = new self($source, $maxDocs);
+		if ($stream->isEmpty()) {
+			return null;
+		}
+		return $stream;
+	}
+
+	/**
+	 * @return bool
+	 */
+	public function isEmpty(): bool {
+		return $this->empty;
+	}
+
+	/**
+	 * @return int
+	 */
+	public function getDocs(): int {
+		return $this->docs;
+	}
+
+	/**
+	 * @return int
+	 */
+	public function getBytes(): int {
+		return $this->bytes;
+	}
+
+	/**
+	 * @return string|null
+	 */
+	public function getTable() {
+		return $this->table;
+	}
+
+	/**
+	 * Pull up to $size encoded bytes for curl's READFUNCTION.
+	 *
+	 * @param int $size
+	 * @return string Empty string signals EOF for this request
+	 */
+	public function read(int $size): string {
+		if ($size <= 0) {
+			return '';
+		}
+		while (strlen($this->buffer) < $size && !$this->closed) {
+			if (!$this->appendNextLine()) {
+				$this->closed = true;
+				break;
+			}
+		}
+		if ($this->buffer === '') {
+			return '';
+		}
+		$chunk = substr($this->buffer, 0, $size);
+		$this->buffer = substr($this->buffer, strlen($chunk));
+		return $chunk;
+	}
+
+	/**
+	 * @return void
+	 */
+	private function prime(): void {
+		if ($this->appendNextLine()) {
+			$this->empty = false;
+		} else {
+			$this->closed = true;
+		}
+	}
+
+	/**
+	 * @return bool true when a line was appended
+	 */
+	private function appendNextLine(): bool {
+		if ($this->maxDocs > 0 && $this->docs >= $this->maxDocs) {
+			return false;
+		}
+		$payload = $this->source->next();
+		if ($payload === null) {
+			return false;
+		}
+
+		if ($this->table !== null && $payload['table'] !== $this->table) {
+			$this->source->unget($payload);
+			return false;
+		}
+
+		if (isset($this->ids[$payload['id']])) {
+			throw new RuntimeException(
+				'Duplicate document id ' . $payload['id'] . ' within the same bulk request'
+			);
+		}
+
+		$line = $this->source->getNormalizer()->encodeLine($payload) . "\n";
+		$this->table = $payload['table'];
+		$this->ids[$payload['id']] = true;
+		$this->buffer .= $line;
+		$this->bytes += strlen($line);
+		$this->docs++;
+		return true;
+	}
+}
diff --git a/src/Manticoresearch/Bulk/OperationIterator.php b/src/Manticoresearch/Bulk/OperationIterator.php
new file mode 100644
index 0000000..ab93ff2
--- /dev/null
+++ b/src/Manticoresearch/Bulk/OperationIterator.php
@@ -0,0 +1,85 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Bulk;
+
+/**
+ * Single-pass cursor over bulk operations with one-slot pushback.
+ *
+ * Concurrent NdjsonStream readers share one cursor; curl_multi invokes
+ * READFUNCTION callbacks on the same PHP thread, so no locking is required.
+ */
+class OperationIterator
+{
+	/** @var OperationNormalizer */
+	private $normalizer;
+
+	/** @var \Iterator */
+	private $inner;
+
+	/** @var array|null */
+	private $pending;
+
+	/** @var bool */
+	private $started = false;
+
+	/**
+	 * @param iterable $operations
+	 * @param OperationNormalizer $normalizer
+	 */
+	public function __construct(iterable $operations, OperationNormalizer $normalizer) {
+		$this->normalizer = $normalizer;
+		if ($operations instanceof \Iterator) {
+			$this->inner = $operations;
+		} elseif ($operations instanceof \Traversable) {
+			$this->inner = new \IteratorIterator($operations);
+		} else {
+			$this->inner = new \ArrayIterator($operations);
+		}
+	}
+
+	/**
+	 * @return OperationNormalizer
+	 */
+	public function getNormalizer(): OperationNormalizer {
+		return $this->normalizer;
+	}
+
+	/**
+	 * @return array|null Normalized insert payload or null at end of input
+	 */
+	public function next() {
+		if ($this->pending !== null) {
+			$payload = $this->pending;
+			$this->pending = null;
+			return $payload;
+		}
+		if (!$this->started) {
+			$this->started = true;
+			$this->inner->rewind();
+		} else {
+			$this->inner->next();
+		}
+		if (!$this->inner->valid()) {
+			return null;
+		}
+		return $this->normalizer->normalize($this->inner->current());
+	}
+
+	/**
+	 * Push one payload back so another stream / the same stream boundary can consume it.
+	 *
+	 * @param array $payload
+	 * @return void
+	 */
+	public function unget(array $payload): void {
+		if ($this->pending !== null) {
+			throw new \LogicException('OperationIterator only supports one pushback slot');
+		}
+		$this->pending = $payload;
+	}
+}
diff --git a/src/Manticoresearch/Bulk/OperationNormalizer.php b/src/Manticoresearch/Bulk/OperationNormalizer.php
new file mode 100644
index 0000000..c1d5956
--- /dev/null
+++ b/src/Manticoresearch/Bulk/OperationNormalizer.php
@@ -0,0 +1,212 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Bulk;
+
+use Manticoresearch\Exceptions\RuntimeException;
+
+/**
+ * Validates and normalizes insert operations for indexer-assisted bulk mode.
+ */
+class OperationNormalizer
+{
+	/** @var string|null */
+	private $expectedTable;
+
+	/** @var bool */
+	private $rejectCluster;
+
+	/**
+	 * @param string|null $expectedTable When set, every operation must target this table
+	 * @param bool $rejectCluster Reject operations that include a cluster field
+	 */
+	public function __construct(?string $expectedTable = null, bool $rejectCluster = true) {
+		$this->expectedTable = $expectedTable;
+		$this->rejectCluster = $rejectCluster;
+	}
+
+	/**
+	 * Normalize one bulk action into an insert payload array.
+	 *
+	 * Accepts either:
+	 * - ['insert' => ['table' => ..., 'id' => ..., 'doc' => ...]]
+	 * - ['table' => ..., 'id' => ..., 'doc' => ...]
+	 * - document arrays with an id key (when expectedTable is set)
+	 *
+	 * @param mixed $operation
+	 * @return array{table:string,id:int|string,doc:array}
+	 */
+	public function normalize($operation): array {
+		if (is_object($operation)) {
+			$operation = (array)$operation;
+		} elseif (is_string($operation)) {
+			$decoded = json_decode($operation, true, 512, JSON_BIGINT_AS_STRING);
+			if (!is_array($decoded)) {
+				throw new RuntimeException('Invalid JSON bulk operation');
+			}
+			$operation = $decoded;
+		}
+		if (!is_array($operation)) {
+			throw new RuntimeException('Bulk operation must be an array or JSON object');
+		}
+
+		if (isset($operation['insert'])) {
+			if (!is_array($operation['insert'])) {
+				throw new RuntimeException('insert action payload must be an array');
+			}
+			$payload = $operation['insert'];
+		} elseif ($this->isActionEnvelope($operation)) {
+			$actions = array_keys($operation);
+			throw new RuntimeException(
+				'indexer-assisted bulk supports insert only; got action: ' . $actions[0]
+			);
+		} else {
+			$payload = $operation;
+		}
+
+		return $this->normalizePayload($payload);
+	}
+
+	/**
+	 * Encode a normalized insert payload as one NDJSON line (without trailing newline).
+	 *
+	 * @param array $payload
+	 * @return string
+	 */
+	public function encodeLine(array $payload): string {
+		$payload['id'] = self::normalizeDocumentId($payload['id']);
+		$flags = JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_PRESERVE_ZERO_FRACTION | JSON_THROW_ON_ERROR;
+		try {
+			if (is_string($payload['id'])) {
+				// Emit validated uint64 digits directly: converting through int or float loses precision.
+				return '{"insert":{"table":' . json_encode($payload['table'], $flags)
+					. ',"id":' . $payload['id']
+					. ',"doc":' . json_encode($payload['doc'], $flags) . '}}';
+			}
+			return json_encode(['insert' => $payload], $flags);
+		} catch (\JsonException $e) {
+			throw new RuntimeException('Failed to JSON-encode bulk insert: ' . $e->getMessage(), 0, $e);
+		}
+	}
+
+	/**
+	 * @param array $payload
+	 * @return array{table:string,id:int|string,doc:array}
+	 */
+	private function normalizePayload(array $payload): array {
+		if ($this->rejectCluster && isset($payload['cluster'])) {
+			throw new RuntimeException(
+				'indexer-assisted bulk does not support clustered/replicated tables'
+			);
+		}
+
+		$table = $payload['table'] ?? $payload['index'] ?? $this->expectedTable;
+		if ($table === null || $table === '') {
+			throw new RuntimeException('Bulk insert requires a table name');
+		}
+		$table = (string)$table;
+		if ($this->expectedTable !== null && $table !== $this->expectedTable) {
+			throw new RuntimeException(
+				'indexer-assisted bulk transaction may target only one table; expected '
+				. $this->expectedTable . ', got ' . $table
+			);
+		}
+
+		if (!array_key_exists('id', $payload)) {
+			throw new RuntimeException(
+				'indexer-assisted bulk requires an explicit non-zero numeric document id'
+			);
+		}
+		$id = self::normalizeDocumentId($payload['id']);
+		if ($id === 0) {
+			throw new RuntimeException(
+				'indexer-assisted bulk requires an explicit non-zero numeric document id'
+			);
+		}
+
+		if (isset($payload['doc'])) {
+			if (!is_array($payload['doc'])) {
+				throw new RuntimeException('insert doc must be an array');
+			}
+			$doc = $payload['doc'];
+		} else {
+			$doc = $payload;
+			unset($doc['id'], $doc['table'], $doc['index'], $doc['cluster']);
+		}
+		$this->rejectNulls($doc);
+
+		return [
+			'table' => $table,
+			'id' => $id,
+			'doc' => $doc,
+		];
+	}
+
+	/**
+	 * Preserve IDs above PHP_INT_MAX as canonical decimal strings.
+	 *
+	 * @param mixed $id
+	 * @return int|string
+	 */
+	private static function normalizeDocumentId($id) {
+		if (is_int($id) && $id >= 0) {
+			return $id;
+		}
+		if (!is_string($id) || preg_match('/\A[0-9]+\z/', $id) !== 1) {
+			throw new RuntimeException('Incorrect document id passed');
+		}
+
+		$id = ltrim($id, '0');
+		if ($id === '') {
+			return 0;
+		}
+
+		$uint64Max = '18446744073709551615';
+		if (strlen($id) > 20
+			|| (strlen($id) === 20 && strcmp($id, $uint64Max) > 0)) {
+			throw new RuntimeException('Document id exceeds uint64 range');
+		}
+
+		$nativeMax = (string)PHP_INT_MAX;
+		if (strlen($id) < strlen($nativeMax)
+			|| (strlen($id) === strlen($nativeMax) && strcmp($id, $nativeMax) <= 0)) {
+			return (int)$id;
+		}
+		return $id;
+	}
+
+	/**
+	 * @param array $operation
+	 * @return bool
+	 */
+	private function isActionEnvelope(array $operation): bool {
+		if (sizeof($operation) !== 1) {
+			return false;
+		}
+		$key = array_keys($operation)[0];
+		return in_array($key, ['update', 'replace', 'delete', 'create', 'index'], true);
+	}
+
+	/**
+	 * @param array $doc
+	 * @param string $prefix
+	 * @return void
+	 */
+	private function rejectNulls(array $doc, string $prefix = ''): void {
+		foreach ($doc as $key => $value) {
+			$path = $prefix === '' ? (string)$key : $prefix . '[' . $key . ']';
+			if ($value === null) {
+				throw new RuntimeException("Error: The key '{$path}' in document has a null value.\n");
+			}
+			if (!is_array($value)) {
+				continue;
+			}
+
+			$this->rejectNulls($value, $path);
+		}
+	}
+}
diff --git a/src/Manticoresearch/Bulk/Options.php b/src/Manticoresearch/Bulk/Options.php
new file mode 100644
index 0000000..cd09936
--- /dev/null
+++ b/src/Manticoresearch/Bulk/Options.php
@@ -0,0 +1,87 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Bulk;
+
+use Manticoresearch\Exceptions\RuntimeException;
+
+/**
+ * Parsed options for indexer-assisted bulk ingestion.
+ */
+class Options
+{
+	/** @var int|null */
+	public $workers;
+
+	/** @var int|null */
+	public $maxWorkers;
+
+	/** @var float Minimum relative throughput gain to keep doubling workers */
+	public $gainThreshold;
+
+	/** @var int|null */
+	public $timeout;
+
+	/** @var bool Fall back to regular Client::bulk() when assisted mode is unavailable */
+	public $fallback;
+
+	/** @var int Documents per worker during each auto-tune probe */
+	public $probeDocs;
+
+	/**
+	 * @param array $options
+	 * @return self
+	 */
+	public static function fromArray(array $options = []): self {
+		$parsed = new self();
+		$parsed->workers = self::optionalPositiveInt($options, 'workers');
+		$parsed->maxWorkers = self::optionalPositiveInt($options, 'max_workers');
+		$parsed->gainThreshold = isset($options['gain_threshold'])
+			? (float)$options['gain_threshold']
+			: 0.0;
+		if ($parsed->gainThreshold < 0) {
+			throw new RuntimeException('gain_threshold must be >= 0');
+		}
+		$parsed->timeout = self::optionalPositiveInt($options, 'timeout');
+		$parsed->fallback = !empty($options['fallback']);
+		$parsed->probeDocs = self::positiveInt($options, 'probe_docs', 1000);
+		return $parsed;
+	}
+
+	/**
+	 * @param array $options
+	 * @param string $key
+	 * @return int|null
+	 */
+	private static function optionalPositiveInt(array $options, string $key) {
+		if (!array_key_exists($key, $options) || $options[$key] === null) {
+			return null;
+		}
+		$value = (int)$options[$key];
+		if ($value < 1) {
+			throw new RuntimeException($key . ' must be >= 1');
+		}
+		return $value;
+	}
+
+	/**
+	 * @param array $options
+	 * @param string $key
+	 * @param int $default
+	 * @return int
+	 */
+	private static function positiveInt(array $options, string $key, int $default): int {
+		if (!array_key_exists($key, $options) || $options[$key] === null) {
+			return $default;
+		}
+		$value = (int)$options[$key];
+		if ($value < 1) {
+			throw new RuntimeException($key . ' must be >= 1');
+		}
+		return $value;
+	}
+}
diff --git a/src/Manticoresearch/Bulk/WorkerCap.php b/src/Manticoresearch/Bulk/WorkerCap.php
new file mode 100644
index 0000000..a42a554
--- /dev/null
+++ b/src/Manticoresearch/Bulk/WorkerCap.php
@@ -0,0 +1,89 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Bulk;
+
+use Manticoresearch\Client;
+
+/**
+ * Resolves a concurrency ceiling from Manticore SHOW STATUS workers_total.
+ */
+class WorkerCap
+{
+	/** @var Client */
+	private $client;
+
+	/**
+	 * @param Client $client
+	 */
+	public function __construct(Client $client) {
+		$this->client = $client;
+	}
+
+	/**
+	 * @return int|null
+	 */
+	public function workersTotal() {
+		try {
+			$rows = $this->client->nodes()->status(['body' => ['pattern' => 'workers_total']]);
+		} catch (\Throwable $e) {
+			return null;
+		}
+		if (!is_array($rows)) {
+			return null;
+		}
+		if (array_key_exists('workers_total', $rows)) {
+			return self::toInt($rows['workers_total']);
+		}
+		foreach ($rows as $row) {
+			if (!is_array($row)) {
+				continue;
+			}
+			$name = $row['Variable_name'] ?? $row['Counter'] ?? $row['Key'] ?? null;
+			if ((string)$name !== 'workers_total') {
+				continue;
+			}
+			return self::toInt($row['Value'] ?? null);
+		}
+		return null;
+	}
+
+	/**
+	 * Default hard concurrency ceiling. Falls back to 1 when status is unavailable.
+	 *
+	 * @param int|null $configuredMax
+	 * @return int
+	 */
+	public function resolveMax(?int $configuredMax = null): int {
+		$serverMax = $this->workersTotal();
+		if ($serverMax === null || $serverMax < 1) {
+			$serverMax = 1;
+		}
+		if ($configuredMax === null) {
+			return $serverMax;
+		}
+		return max(1, min($configuredMax, $serverMax));
+	}
+
+	/**
+	 * @param mixed $value
+	 * @return int|null
+	 */
+	private static function toInt($value) {
+		if (is_array($value)) {
+			if (isset($value['Value'])) {
+				$value = $value['Value'];
+			} else {
+				$value = reset($value);
+			}
+		}
+		if ($value === null || $value === false || $value === '') {
+			return null;
+		}
+		return (int)$value;
+	}
+}
diff --git a/src/Manticoresearch/Client.php b/src/Manticoresearch/Client.php
index f0301bb..df0df6f 100755
--- a/src/Manticoresearch/Client.php
+++ b/src/Manticoresearch/Client.php
@@ -357,6 +357,30 @@ public function bulk(array $params = []) {
 		return $response->getResponse();
 	}
 
+	/**
+	 * Indexer-assisted bulk insert via /bulk?indexer_rt_bulk=1.
+	 *
+	 * Experimental server feature: streams insert-only NDJSON, optionally with
+	 * concurrent requests that drain a shared iterator until empty. When
+	 * `workers` is omitted, concurrency is probed as 1,2,4,... up to the
+	 * server workers_total (or max_workers). `probe_docs` only affects those
+	 * auto-tune probes.
+	 *
+	 * @param iterable $operations Insert operations or documents
+	 * @param array $options workers, max_workers, probe_docs, gain_threshold,
+	 *                       timeout, fallback
+	 * @return array
+	 */
+	public function indexerBulk(iterable $operations, array $options = []): array {
+		$normalizer = null;
+		if (isset($options['normalizer']) && $options['normalizer'] instanceof Bulk\OperationNormalizer) {
+			$normalizer = $options['normalizer'];
+			unset($options['normalizer']);
+		}
+		$runner = new Bulk\IndexerBulk($this, $this->logger, $normalizer);
+		return $runner->run($operations, $options);
+	}
+
 	/**
 	 * Endpoint: suggest
 	 * @param array $params
diff --git a/src/Manticoresearch/ClientInterface.php b/src/Manticoresearch/ClientInterface.php
index dff70bb..837bc09 100644
--- a/src/Manticoresearch/ClientInterface.php
+++ b/src/Manticoresearch/ClientInterface.php
@@ -155,6 +155,14 @@ public function table(?string $name = null): Table;
 	 */
 	public function bulk(array $params = []);
 
+	/**
+	 * Indexer-assisted bulk insert via /bulk?indexer_rt_bulk=1
+	 * @param iterable $operations
+	 * @param array $options
+	 * @return array
+	 */
+	public function indexerBulk(iterable $operations, array $options = []): array;
+
 	/**
 	 * Endpoint: suggest
 	 * @param array $params
diff --git a/src/Manticoresearch/Table.php b/src/Manticoresearch/Table.php
index f7c9440..9441f9b 100755
--- a/src/Manticoresearch/Table.php
+++ b/src/Manticoresearch/Table.php
@@ -125,6 +125,30 @@ public function addDocuments($documents) {
 		return $this->client->bulk(['body' => $toinsert]);
 	}
 
+	/**
+	 * Fast append-only bulk insert using indexer-assisted mode when available.
+	 *
+	 * Requires explicit non-zero numeric ids. Clustered tables are not supported.
+	 *
+	 * @param iterable $documents
+	 * @param array $options See Client::indexerBulk()
+	 * @return array
+	 */
+	public function addDocumentsStreaming($documents, array $options = []): array {
+		if ($this->cluster !== null) {
+			throw new RuntimeException(
+				'addDocumentsStreaming does not support clustered/replicated tables'
+			);
+		}
+		if ($this->table === null || $this->table === '') {
+			throw new RuntimeException('Table name is required for addDocumentsStreaming');
+		}
+
+		$normalizer = new Bulk\OperationNormalizer($this->table, true);
+		$options['normalizer'] = $normalizer;
+		return $this->client->indexerBulk($documents, $options);
+	}
+
 	public function deleteDocument($id) {
 		static::checkDocumentId($id);
 		$params = [
diff --git a/src/Manticoresearch/Transport.php b/src/Manticoresearch/Transport.php
index 47dc11d..ff8a6d1 100644
--- a/src/Manticoresearch/Transport.php
+++ b/src/Manticoresearch/Transport.php
@@ -128,7 +128,7 @@ protected function buildAuthorizationHeader(Connection $connection): ?string {
 	 * @param Connection $connection
 	 * @return array
 	 */
-	protected function getRequestHeadersAsList(Request $request, Connection $connection): array {
+	public function getRequestHeadersAsList(Request $request, Connection $connection): array {
 		$headers = $connection->getHeaders();
 		$headers[] = sprintf('Content-Type: %s', $request->getContentType());
 		$authorization = $this->buildAuthorizationHeader($connection);
diff --git a/test/Manticoresearch/Bulk/IndexerBulkIntegrationTest.php b/test/Manticoresearch/Bulk/IndexerBulkIntegrationTest.php
new file mode 100644
index 0000000..7255e10
--- /dev/null
+++ b/test/Manticoresearch/Bulk/IndexerBulkIntegrationTest.php
@@ -0,0 +1,97 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Test\Bulk;
+
+use Manticoresearch\Client;
+use Manticoresearch\Exceptions\RuntimeException;
+use Manticoresearch\Test\Helper\PopulateHelperTest;
+use PHPUnit\Framework\TestCase;
+
+/**
+ * Live tests for indexer-assisted bulk. Skipped when the server build lacks the feature.
+ */
+class IndexerBulkIntegrationTest extends TestCase
+{
+	/** @var Client */
+	private static $client;
+
+	public static function setUpBeforeClass(): void {
+		parent::setUpBeforeClass();
+		$helper = new PopulateHelperTest('testDummy');
+		static::$client = $helper->getClient();
+	}
+
+	public function testAddDocumentsFastOrSkip() {
+		$tableName = 'indexer_bulk_php_' . getmypid();
+		$client = static::$client;
+		$client->tables()->drop(['table' => $tableName, 'body' => ['silent' => true]]);
+		$client->tables()->create(
+			[
+				'table' => $tableName,
+				'body' => [
+					'columns' => [
+						'title' => ['type' => 'text'],
+						'gid' => ['type' => 'integer'],
+					],
+				],
+			]
+		);
+
+		$table = $client->table($tableName);
+		$docs = [];
+		for ($i = 1; $i <= 50; $i++) {
+			$docs[] = ['id' => $i, 'title' => 'doc ' . $i, 'gid' => $i % 5];
+		}
+
+		try {
+			$result = $table->addDocumentsStreaming(
+				$docs,
+				[
+					'workers' => 2,
+				]
+			);
+		} catch (RuntimeException $e) {
+			$client->tables()->drop(['table' => $tableName, 'body' => ['silent' => true]]);
+			$this->markTestSkipped('indexer_rt_bulk unavailable: ' . $e->getMessage());
+			return;
+		} catch (\Manticoresearch\Exceptions\ResponseException $e) {
+			$client->tables()->drop(['table' => $tableName, 'body' => ['silent' => true]]);
+			$this->markTestSkipped('indexer_rt_bulk unavailable: ' . $e->getMessage());
+			return;
+		}
+
+		$this->assertSame('indexer_rt_bulk', $result['mode']);
+		$this->assertSame(50, $result['docs']);
+		$this->assertGreaterThanOrEqual(1, $result['requests']);
+
+		// With limit=0, recent Manticore returns total=0 (total_relation=gte).
+		$search = $client->search(
+			[
+				'body' => [
+					'table' => $tableName,
+					'query' => ['match_all' => ''],
+					'limit' => 100,
+				],
+			]
+		);
+		$this->assertSame(50, $search['hits']['total']);
+
+		$client->tables()->drop(['table' => $tableName, 'body' => ['silent' => true]]);
+	}
+
+	public function testRejectsMissingIds() {
+		$table = static::$client->table('indexer_bulk_php_invalid');
+		$this->expectException(RuntimeException::class);
+		$table->addDocumentsStreaming(
+			[
+				['title' => 'no id'],
+			],
+			['workers' => 1]
+		);
+	}
+}
diff --git a/test/Manticoresearch/Bulk/IndexerBulkUnitTest.php b/test/Manticoresearch/Bulk/IndexerBulkUnitTest.php
new file mode 100644
index 0000000..4e66df9
--- /dev/null
+++ b/test/Manticoresearch/Bulk/IndexerBulkUnitTest.php
@@ -0,0 +1,390 @@
+<?php
+
+// Copyright (c) Manticore Software LTD (https://manticoresearch.com)
+//
+// This source code is licensed under the MIT license found in the
+// LICENSE file in the root directory of this source tree.
+
+namespace Manticoresearch\Test\Bulk;
+
+use Manticoresearch\Bulk\AdaptiveTuner;
+use Manticoresearch\Bulk\IndexerBulk;
+use Manticoresearch\Bulk\NdjsonStream;
+use Manticoresearch\Bulk\OperationIterator;
+use Manticoresearch\Bulk\OperationNormalizer;
+use Manticoresearch\Bulk\Options;
+use Manticoresearch\Client;
+use Manticoresearch\Exceptions\RuntimeException;
+use PHPUnit\Framework\TestCase;
+
+class IndexerBulkUnitTest extends TestCase
+{
+	public function testOptionsDefaults() {
+		$opts = Options::fromArray([]);
+		$this->assertNull($opts->workers);
+		$this->assertSame(1000, $opts->probeDocs);
+		$this->assertSame(0.0, $opts->gainThreshold);
+		$this->assertFalse($opts->fallback);
+	}
+
+	public function testOptionsRejectInvalidWorkers() {
+		$this->expectException(RuntimeException::class);
+		Options::fromArray(['workers' => 0]);
+	}
+
+	public function testNormalizeInsertEnvelope() {
+		$normalizer = new OperationNormalizer();
+		$payload = $normalizer->normalize(
+			[
+				'insert' => [
+					'table' => 'products',
+					'id' => 10,
+					'doc' => ['title' => 'Bag'],
+				],
+			]
+		);
+		$this->assertSame(
+			['table' => 'products', 'id' => 10, 'doc' => ['title' => 'Bag']],
+			$payload
+		);
+	}
+
+	public function testNormalizeDocumentWithExpectedTable() {
+		$normalizer = new OperationNormalizer('products');
+		$payload = $normalizer->normalize(['id' => 3, 'title' => 'Bag']);
+		$this->assertSame(3, $payload['id']);
+		$this->assertSame('products', $payload['table']);
+		$this->assertSame(['title' => 'Bag'], $payload['doc']);
+	}
+
+	public function testRejectZeroId() {
+		$normalizer = new OperationNormalizer('products');
+		$this->expectException(RuntimeException::class);
+		$normalizer->normalize(['id' => 0, 'title' => 'x']);
+	}
+
+	/** @dataProvider validDocumentIds */
+	#[\PHPUnit\Framework\Attributes\DataProvider('validDocumentIds')]
+	public function testNormalizeDocumentIdWithoutPrecisionLoss($id, $expected) {
+		$normalizer = new OperationNormalizer('products');
+		$payload = $normalizer->normalize(['id' => $id, 'title' => 'x']);
+		$this->assertSame($expected, $payload['id']);
+	}
+
+	public static function validDocumentIds(): array {
+		return [
+			'native integer' => [42, 42],
+			'leading zeros' => ['00042', 42],
+			'native maximum integer' => [PHP_INT_MAX, PHP_INT_MAX],
+			'native maximum string' => [(string)PHP_INT_MAX, PHP_INT_MAX],
+			'above signed 64-bit range' => ['9223372036854775808', '9223372036854775808'],
+			'unsigned maximum' => ['18446744073709551615', '18446744073709551615'],
+			'unsigned maximum with leading zeros' => ['0018446744073709551615', '18446744073709551615'],
+		];
+	}
+
+	/** @dataProvider invalidDocumentIds */
+	#[\PHPUnit\Framework\Attributes\DataProvider('invalidDocumentIds')]
+	public function testRejectInvalidDocumentId($id) {
+		$normalizer = new OperationNormalizer('products');
+		$this->expectException(RuntimeException::class);
+		$normalizer->normalize(['id' => $id, 'title' => 'x']);
+	}
+
+	public static function invalidDocumentIds(): array {
+		return [
+			'zero string' => ['000'],
+			'overflow' => ['18446744073709551616'],
+			'long overflow' => ['100000000000000000000'],
+			'negative integer' => [-1],
+			'integral float' => [1.0],
+			'fractional float' => [1.5],
+			'decimal string' => ['1.5'],
+			'exponent string' => ['1e3'],
+			'negative string' => ['-42'],
+			'positive sign' => ['+42'],
+			'leading whitespace' => [' 42'],
+			'trailing whitespace' => ['42 '],
+			'trailing newline' => ["42\n"],
+			'empty string' => [''],
+			'non-ASCII digits' => ['１２'],
+			'boolean' => [true],
+			'null' => [null],
+			'array' => [[1]],
+			'object' => [(object)['id' => 1]],
+		];
+	}
+
+	public function testNormalizeJsonPreservesUnsignedDocumentId() {
+		$normalizer = new OperationNormalizer();
+		$payload = $normalizer->normalize(
+			'{"insert":{"table":"products","id":18446744073709551615,"doc":{"title":"x"}}}'
+		);
+		$this->assertSame('18446744073709551615', $payload['id']);
+	}
+
+	public function testRejectNonInsertAction() {
+		$normalizer = new OperationNormalizer();
+		$this->expectException(RuntimeException::class);
+		$normalizer->normalize(['delete' => ['table' => 'products', 'id' => 1]]);
+	}
+
+	public function testRejectCluster() {
+		$normalizer = new OperationNormalizer('products');
+		$this->expectException(RuntimeException::class);
+		$normalizer->normalize(
+			[
+				'insert' => [
+					'table' => 'products',
+					'cluster' => 'c1',
+					'id' => 1,
+					'doc' => ['title' => 'x'],
+				],
+			]
+		);
+	}
+
+	public function testEncodeLine() {
+		$normalizer = new OperationNormalizer();
+		$line = $normalizer->encodeLine(
+			['table' => 'products', 'id' => 1, 'doc' => ['title' => 'x']]
+		);
+		$this->assertSame(
+			'{"insert":{"table":"products","id":1,"doc":{"title":"x"}}}',
+			$line
+		);
+		$this->assertStringNotContainsString("\n", $line);
+	}
+
+	public function testEncodeLinePreservesUnsignedIdAndDocumentTypes() {
+		$normalizer = new OperationNormalizer();
+		$payload = [
+			'table' => "products\"\\\n",
+			'id' => '18446744073709551615',
+			'doc' => [
+				'id' => '18446744073709551615',
+				'nested' => ['id' => '9223372036854775808'],
+				'code' => '00123',
+				'price' => 1.0,
+				'title' => "Bag\n雪/",
+			],
+		];
+		$line = $normalizer->encodeLine($payload);
+		$this->assertStringContainsString('"id":18446744073709551615,', $line);
+		$this->assertStringContainsString('"price":1.0', $line);
+		$this->assertStringNotContainsString("\n", $line);
+		$this->assertSame(['insert' => $payload], json_decode($line, true, 512, JSON_BIGINT_AS_STRING));
+	}
+
+	public function testEncodeLineRejectsInvalidIdBeforeRawEmission() {
+		$normalizer = new OperationNormalizer();
+		$this->expectException(RuntimeException::class);
+		$normalizer->encodeLine(['table' => 'products', 'id' => '1,"extra":true', 'doc' => []]);
+	}
+
+	public function testEncodeLineReportsJsonEncodingFailureForUnsignedId() {
+		$normalizer = new OperationNormalizer();
+		$this->expectException(RuntimeException::class);
+		$this->expectExceptionMessage('Failed to JSON-encode bulk insert:');
+		$normalizer->encodeLine(
+			['table' => 'products', 'id' => '18446744073709551615', 'doc' => ['title' => "\xB1\x31"]]
+		);
+	}
+
+	public function testFallbackBulkPreservesUnsignedIdAsJsonNumber() {
+		$client = $this->createMock(Client::class);
+		$client->expects($this->once())->method('getConnectionPool')
+			->willThrowException(new RuntimeException('Unsupported transport'));
+		$body = '{"insert":{"table":"t","id":1,"doc":{"title":"a"}}}' . "\n"
+			. '{"insert":{"table":"t","id":18446744073709551615,"doc":{"title":"b"}}}' . "\n";
+		$client->expects($this->once())->method('bulk')->with(['body' => $body])->willReturn(['items' => []]);
+		$indexer = new IndexerBulk($client);
+		$result = $indexer->run(
+			[
+				['insert' => ['table' => 't', 'id' => 1, 'doc' => ['title' => 'a']]],
+				['table' => 't', 'id' => '18446744073709551615', 'doc' => ['title' => 'b']],
+			],
+			['fallback' => true]
+		);
+		$this->assertSame(2, $result['docs']);
+		$this->assertSame('bulk_fallback', $result['mode']);
+	}
+
+	public function testNdjsonStreamSplitsByDocLimit() {
+		$normalizer = new OperationNormalizer('t');
+		$source = new OperationIterator(
+			[
+				['id' => 1, 'title' => 'a'],
+				['id' => 2, 'title' => 'b'],
+				['id' => 3, 'title' => 'c'],
+			],
+			$normalizer
+		);
+
+		$first = NdjsonStream::tryCreate($source, 2);
+		$this->assertNotNull($first);
+		$body = '';
+		while (($chunk = $first->read(64)) !== '') {
+			$body .= $chunk;
+		}
+		$this->assertSame(2, $first->getDocs());
+		$this->assertStringEndsWith("\n", $body);
+		$this->assertStringNotContainsString("\n\n", $body);
+
+		$second = NdjsonStream::tryCreate($source, 2);
+		$this->assertNotNull($second);
+		while ($second->read(64) !== '') {
+			// drain
+		}
+		$this->assertSame(1, $second->getDocs());
+
+		$this->assertNull(NdjsonStream::tryCreate($source, 2));
+	}
+
+	public function testWorkersDrainSharedSourceUntilEmpty() {
+		$normalizer = new OperationNormalizer('t');
+		$ops = [];
+		for ($i = 1; $i <= 10; $i++) {
+			$ops[] = ['id' => $i, 'title' => 'd' . $i];
+		}
+		$source = new OperationIterator($ops, $normalizer);
+
+		// Two unlimited streams share the iterator (same model as final upload).
+		$streams = [];
+		for ($i = 0; $i < 2; $i++) {
+			$stream = NdjsonStream::tryCreate($source);
+			$this->assertNotNull($stream);
+			$streams[] = $stream;
+		}
+
+		$docs = 0;
+		foreach ($streams as $stream) {
+			while ($stream->read(128) !== '') {
+				// drain
+			}
+			$docs += $stream->getDocs();
+		}
+		$this->assertSame(10, $docs);
+		$this->assertNull(NdjsonStream::tryCreate($source));
+	}
+
+	public function testNdjsonStreamEncodesIncrementally() {
+		$normalizer = new OperationNormalizer('t');
+		$source = new OperationIterator(
+			[
+				['id' => 1, 'title' => 'a'],
+				['id' => 2, 'title' => 'b'],
+			],
+			$normalizer
+		);
+		$stream = NdjsonStream::tryCreate($source, 10);
+		$this->assertNotNull($stream);
+		// First primed line is buffered; subsequent docs are encoded only as read() needs bytes.
+		$firstChunk = $stream->read(1);
+		$this->assertSame(1, strlen($firstChunk));
+		$this->assertSame(1, $stream->getDocs());
+		while ($stream->read(1024) !== '') {
+			// drain remaining including second document
+		}
+		$this->assertSame(2, $stream->getDocs());
+	}
+
+	public function testNdjsonStreamRejectsDuplicateIdsInRequest() {
+		$normalizer = new OperationNormalizer('t');
+		$source = new OperationIterator(
+			[
+				['id' => 1, 'title' => 'a'],
+				['id' => 1, 'title' => 'b'],
+			],
+			$normalizer
+		);
+		$stream = NdjsonStream::tryCreate($source, 10);
+		$this->assertNotNull($stream);
+		$this->expectException(RuntimeException::class);
+		while ($stream->read(1024) !== '') {
+			// drain until duplicate is pulled
+		}
+	}
+
+	public function testNdjsonStreamPreservesDistinctUnsignedIds() {
+		$normalizer = new OperationNormalizer('t');
+		$source = new OperationIterator(
+			[
+				['id' => '9223372036854775808', 'title' => 'a'],
+				['id' => '18446744073709551615', 'title' => 'b'],
+			],
+			$normalizer
+		);
+		$stream = NdjsonStream::tryCreate($source);
+		$this->assertNotNull($stream);
+		$body = '';
+		while (($chunk = $stream->read(64)) !== '') {
+			$body .= $chunk;
+		}
+		$this->assertSame(2, $stream->getDocs());
+		$this->assertSame(
+			'{"insert":{"table":"t","id":9223372036854775808,"doc":{"title":"a"}}}' . "\n"
+			. '{"insert":{"table":"t","id":18446744073709551615,"doc":{"title":"b"}}}' . "\n",
+			$body
+		);
+	}
+
+	/** @dataProvider duplicateDocumentIds */
+	#[\PHPUnit\Framework\Attributes\DataProvider('duplicateDocumentIds')]
+	public function testNdjsonStreamRejectsCanonicalDuplicateIds($firstId, $secondId) {
+		$normalizer = new OperationNormalizer('t');
+		$source = new OperationIterator(
+			[
+				['id' => $firstId, 'title' => 'a'],
+				['id' => $secondId, 'title' => 'b'],
+			],
+			$normalizer
+		);
+		$stream = NdjsonStream::tryCreate($source);
+		$this->assertNotNull($stream);
+		$this->expectException(RuntimeException::class);
+		$this->expectExceptionMessage('Duplicate document id');
+		while ($stream->read(1024) !== '') {
+			// drain until duplicate is pulled
+		}
+	}
+
+	public static function duplicateDocumentIds(): array {
+		return [
+			'native integer and padded string' => [42, '00042'],
+			'unsigned integer and padded string' => ['18446744073709551615', '0018446744073709551615'],
+		];
+	}
+
+	public function testAdaptiveTunerCandidates() {
+		$tuner = new AdaptiveTuner(0.05, 10);
+		$this->assertSame([1, 2, 4, 8, 10], $tuner->candidates());
+	}
+
+	public function testAdaptiveTunerStopsWithoutGain() {
+		$tuner = new AdaptiveTuner(0.10, 8);
+		$calls = [];
+		$result = $tuner->tune(
+			static function ($workers) use (&$calls) {
+				$calls[] = $workers;
+				// 1 -> 100, 2 -> 105 (<10% gain), should stop before 4
+				$map = [1 => 100.0, 2 => 105.0, 4 => 200.0, 8 => 400.0];
+				return $map[$workers];
+			}
+		);
+		$this->assertSame([1, 2], $calls);
+		$this->assertSame(2, $result['selected_workers']);
+	}
+
+	public function testAdaptiveTunerSelectsBestOnImprovement() {
+		$tuner = new AdaptiveTuner(0.05, 4);
+		$result = $tuner->tune(
+			static function ($workers) {
+				$map = [1 => 100.0, 2 => 180.0, 4 => 300.0];
+				return $map[$workers];
+			}
+		);
+		$this->assertSame(4, $result['selected_workers']);
+		$this->assertCount(3, $result['stages']);
+	}
+}