diff --git a/inc/Core/Database/Jobs/Jobs.php b/inc/Core/Database/Jobs/Jobs.php
index 6ad5dc280..c19c04c53 100644
--- a/inc/Core/Database/Jobs/Jobs.php
+++ b/inc/Core/Database/Jobs/Jobs.php
@@ -33,12 +33,15 @@ class Jobs extends BaseRepository {
 
 	const TABLE_NAME = 'datamachine_jobs';
 
-	private const TERMINAL_ACCOUNTING_METRICS   = 0;
-	private const TERMINAL_ACCOUNTING_CORE      = 1;
-	private const TERMINAL_ACCOUNTING_LIFECYCLE = 2;
-	private const TERMINAL_ACCOUNTING_NOTIFY    = 3;
-	public const TERMINAL_ACCOUNTING_COMPLETE   = 4;
-	private const RECOVERY_LEASE_TTL            = 300;
+	private const TERMINAL_ACCOUNTING_METRICS    = 0;
+	private const TERMINAL_ACCOUNTING_CORE       = 1;
+	private const TERMINAL_ACCOUNTING_LIFECYCLE  = 2;
+	private const TERMINAL_ACCOUNTING_NOTIFY     = 3;
+	public const TERMINAL_ACCOUNTING_COMPLETE    = 4;
+	private const RECOVERY_LEASE_TTL             = 300;
+	private const IDEMPOTENT_INSERT_MAX_ATTEMPTS = 3;
+	private const IDEMPOTENT_INSERT_RETRY_US     = 10000;
+	private const INSERT_FAILURE_LOG_TTL         = 300;
 
 	/** Job currently owning the connection-wide terminal transaction. */
 	private static ?int $terminalizing_job = null;
@@ -191,37 +194,38 @@ public function create_or_get_job( array $job_data ): array|false {
 			return false;
 		}
 
-		$inserted = $this->insert_prepared_job( $prepared );
-		if ( false === $inserted ) {
+		$db_error = '';
+		$attempts = 0;
+		while ( $attempts < self::IDEMPOTENT_INSERT_MAX_ATTEMPTS ) {
+			++$attempts;
+			$inserted = $this->insert_prepared_job( $prepared );
+			if ( false !== $inserted ) {
+				break;
+			}
+
+			// Preserve the write error before the recovery read resets wpdb::last_error.
+			$db_error = (string) $this->wpdb->last_error;
 			$existing = $this->get_job_by_idempotency_key( $idempotency_key );
 			if ( null !== $existing ) {
-				( new RunLifecycleStore( $this ) )->mark_job_created(
-					(int) $existing['job_id'],
-					array(
-						'run_type' => $existing['source'] ?? 'job',
-						'status'   => $existing['status'] ?? JobStatus::PENDING,
-					)
-				);
+				return $this->existing_idempotent_job_result( $existing );
+			}
 
-				return array(
-					'job_id'         => (int) $existing['job_id'],
-					'created'        => false,
-					'already_exists' => true,
-					'job'            => $existing,
-				);
+			if ( $this->is_duplicate_database_error( $db_error ) ) {
+				$existing = $this->wait_for_idempotent_race_winner( $idempotency_key );
+				if ( null !== $existing ) {
+					return $this->existing_idempotent_job_result( $existing );
+				}
+				break;
 			}
 
-			do_action(
-				'datamachine_log',
-				'error',
-				'Failed to insert idempotent job',
-				array(
-					'pipeline_id'     => $prepared['pipeline_id'],
-					'flow_id'         => $prepared['flow_id'],
-					'idempotency_key' => $idempotency_key,
-					'db_error'        => $this->wpdb->last_error,
-				)
-			);
+			if ( ! $this->is_retryable_database_error( $db_error ) || $attempts >= self::IDEMPOTENT_INSERT_MAX_ATTEMPTS ) {
+				break;
+			}
+			usleep( self::IDEMPOTENT_INSERT_RETRY_US * $attempts );
+		}
+
+		if ( false === $inserted ) {
+			$this->report_idempotent_insert_failure( $prepared, $idempotency_key, $db_error, $attempts );
 			return false;
 		}
 
@@ -318,7 +322,7 @@ private function prepare_job_insert( array $job_data ): array|false {
 		$default_source = $is_contextless ? 'direct' : ( $is_direct_execution ? 'direct' : 'pipeline' );
 		$source         = sanitize_key( $job_data['source'] ?? $default_source );
 
-		$label = isset( $job_data['label'] ) ? sanitize_text_field( $job_data['label'] ) : null;
+		$label = isset( $job_data['label'] ) ? mb_substr( sanitize_text_field( $job_data['label'] ), 0, 255, 'UTF-8' ) : null;
 
 		$parent_job_id = isset( $job_data['parent_job_id'] ) ? absint( $job_data['parent_job_id'] ) : 0;
 		$user_id       = isset( $job_data['user_id'] ) ? absint( $job_data['user_id'] ) : 0;
@@ -684,6 +688,73 @@ private function insert_prepared_job( array $prepared ): int|false {
 		return $this->wpdb->insert( $this->table_name, $prepared['data'], $prepared['format'] );
 	}
 
+	/** Build the canonical response for a competing idempotent insert winner. */
+	private function existing_idempotent_job_result( array $existing ): array {
+		( new RunLifecycleStore( $this ) )->mark_job_created(
+			(int) $existing['job_id'],
+			array(
+				'run_type' => $existing['source'] ?? 'job',
+				'status'   => $existing['status'] ?? JobStatus::PENDING,
+			)
+		);
+
+		return array(
+			'job_id'         => (int) $existing['job_id'],
+			'created'        => false,
+			'already_exists' => true,
+			'job'            => $existing,
+		);
+	}
+
+	/** Give a committed duplicate-race winner a bounded visibility window. */
+	private function wait_for_idempotent_race_winner( string $idempotency_key ): ?array {
+		for ( $attempt = 1; $attempt < self::IDEMPOTENT_INSERT_MAX_ATTEMPTS; ++$attempt ) {
+			usleep( self::IDEMPOTENT_INSERT_RETRY_US * $attempt );
+			$existing = $this->get_job_by_idempotency_key( $idempotency_key );
+			if ( null !== $existing ) {
+				return $existing;
+			}
+		}
+
+		return null;
+	}
+
+	/** Whether an insert error proves another writer won the unique-key race. */
+	private function is_duplicate_database_error( string $db_error ): bool {
+		return str_contains( $db_error, '1062' ) || false !== stripos( $db_error, 'Duplicate entry' );
+	}
+
+	/** Retry only bounded transaction-contention failures. */
+	private function is_retryable_database_error( string $db_error ): bool {
+		return str_contains( $db_error, '1205' )
+			|| str_contains( $db_error, '1213' )
+			|| false !== stripos( $db_error, 'Lock wait timeout' )
+			|| false !== stripos( $db_error, 'Deadlock found' );
+	}
+
+	/** Emit safe, rate-limited diagnostics while keeping a per-failure signal. */
+	private function report_idempotent_insert_failure( array $prepared, string $idempotency_key, string $db_error, int $attempts ): void {
+		$error_type = $this->is_duplicate_database_error( $db_error ) ? 'duplicate_without_winner' : ( $this->is_retryable_database_error( $db_error ) ? 'contention_exhausted' : 'non_retryable' );
+		$key_type   = str_contains( $idempotency_key, ':' ) ? strstr( $idempotency_key, ':', true ) : 'untyped';
+		$context    = array(
+			'pipeline_id'          => $prepared['pipeline_id'],
+			'flow_id'              => $prepared['flow_id'],
+			'idempotency_key_type' => sanitize_key( $key_type ),
+			'idempotency_key_hash' => hash( 'sha256', $idempotency_key ),
+			'db_error'             => mb_substr( sanitize_text_field( $db_error ), 0, 500, 'UTF-8' ),
+			'error_type'           => $error_type,
+			'attempts'             => $attempts,
+			'log_throttle_seconds' => self::INSERT_FAILURE_LOG_TTL,
+		);
+
+		do_action( 'datamachine_idempotent_job_insert_failed', $context );
+
+		$throttle_key = 'insert_failure_' . hash( 'sha256', $error_type . '|' . $prepared['pipeline_id'] . '|' . $prepared['flow_id'] . '|' . $idempotency_key );
+		if ( wp_cache_add( $throttle_key, 1, 'datamachine_job_insert_failures', self::INSERT_FAILURE_LOG_TTL ) ) {
+			do_action( 'datamachine_log', 'error', 'Failed to insert idempotent job', $context );
+		}
+	}
+
 	/**
 	 * Normalize a job idempotency key for storage and lookup.
 	 *
diff --git a/tests/jobs-idempotent-insert-recovery-smoke.php b/tests/jobs-idempotent-insert-recovery-smoke.php
new file mode 100644
index 000000000..365286e83
--- /dev/null
+++ b/tests/jobs-idempotent-insert-recovery-smoke.php
@@ -0,0 +1,198 @@
+<?php
+/**
+ * Regression coverage for bounded idempotent job insertion recovery.
+ *
+ * Run with: php tests/jobs-idempotent-insert-recovery-smoke.php
+ */
+
+namespace DataMachine\Core {
+	class JobStatus {
+		public const PENDING = 'pending';
+	}
+
+	class RunLifecycleStore {
+		public function __construct( mixed $jobs = null ) {}
+
+		public function mark_job_created( int $job_id, array $seed = array() ): void {}
+	}
+}
+
+namespace {
+	define( 'ABSPATH', __DIR__ . '/' );
+	define( 'ARRAY_A', 'ARRAY_A' );
+	define( 'OBJECT', 'OBJECT' );
+
+	$GLOBALS['job_insert_test_actions'] = array();
+	$GLOBALS['job_insert_test_cache']   = array();
+
+	function absint( mixed $value ): int {
+		return abs( (int) $value );
+	}
+
+	function sanitize_key( mixed $value ): string {
+		return preg_replace( '/[^a-z0-9_\-]/', '', strtolower( (string) $value ) );
+	}
+
+	function sanitize_text_field( mixed $value ): string {
+		return trim( strip_tags( (string) $value ) );
+	}
+
+	function wp_json_encode( mixed $value ): string|false {
+		return json_encode( $value );
+	}
+
+	function current_time( string $type, bool $gmt = false ): string {
+		return '2026-08-06 15:00:00';
+	}
+
+	function do_action( string $hook, mixed ...$args ): void {
+		$GLOBALS['job_insert_test_actions'][ $hook ][] = $args;
+	}
+
+	function wp_cache_add( string $key, mixed $value, string $group = '', int $expire = 0 ): bool {
+		if ( isset( $GLOBALS['job_insert_test_cache'][ $group ][ $key ] ) ) {
+			return false;
+		}
+		$GLOBALS['job_insert_test_cache'][ $group ][ $key ] = $value;
+		return true;
+	}
+
+	class wpdb {
+		public string $prefix = 'wp_';
+		public int $insert_id = 0;
+		public string $last_error = '';
+	}
+
+	class Jobs_Idempotent_Insert_Test_Wpdb extends wpdb {
+		public string $mode = 'success';
+		public int $insert_calls = 0;
+		public int $lookup_calls = 0;
+		/** @var array<int,array<string,mixed>> */
+		public array $rows = array();
+
+		public function insert( mixed $table, array $data, array $format ): int|false {
+			++$this->insert_calls;
+			if ( 'duplicate_race' === $this->mode ) {
+				$this->last_error = "Duplicate entry 'redacted' for key 'idx_idempotency_key' (1062)";
+				return false;
+			}
+			if ( 'contention' === $this->mode ) {
+				$this->last_error = 'Deadlock found when trying to get lock; try restarting transaction (1213)';
+				return false;
+			}
+			if ( 'contention_then_success' === $this->mode && $this->insert_calls < 3 ) {
+				$this->last_error = 'Lock wait timeout exceeded; try restarting transaction (1205)';
+				return false;
+			}
+			if ( 'non_retryable' === $this->mode ) {
+				$this->last_error = "Data too long for column 'operation_step_id' at row 1";
+				return false;
+			}
+
+			$this->last_error = '';
+			$this->insert_id  = count( $this->rows ) + 1;
+			$this->rows[ $this->insert_id ] = array_merge(
+				array(
+					'job_id'     => $this->insert_id,
+					'created_at' => '2026-08-06 15:00:00',
+				),
+				$data
+			);
+			return 1;
+		}
+
+		public function prepare( string $query, mixed ...$args ): array {
+			return array( $query, $args );
+		}
+
+		public function get_row( mixed $query, mixed $output = OBJECT ): ?array {
+			$args = is_array( $query ) ? $query[1] : array();
+			if ( str_contains( (string) $query[0], 'idempotency_key' ) ) {
+				++$this->lookup_calls;
+				$key = (string) end( $args );
+				if ( 'duplicate_race' === $this->mode && $this->lookup_calls >= 3 ) {
+					return array(
+						'job_id'         => 77,
+						'idempotency_key' => $key,
+						'source'          => 'pipeline',
+						'status'          => 'pending',
+					);
+				}
+				foreach ( $this->rows as $row ) {
+					if ( $key === (string) ( $row['idempotency_key'] ?? '' ) ) {
+						return $row;
+					}
+				}
+				return null;
+			}
+
+			$job_id = (int) end( $args );
+			return $this->rows[ $job_id ] ?? null;
+		}
+	}
+
+	require_once dirname( __DIR__ ) . '/inc/Core/Database/BaseRepository.php';
+	require_once dirname( __DIR__ ) . '/inc/Core/Database/Jobs/Jobs.php';
+
+	use DataMachine\Core\Database\Jobs\Jobs;
+
+	$assertions = 0;
+	$assert     = static function ( bool $condition, string $message ) use ( &$assertions ): void {
+		++$assertions;
+		if ( ! $condition ) {
+			throw new RuntimeException( "Assertion failed: {$message}" );
+		}
+	};
+	$new_jobs   = static function ( string $mode ): array {
+		global $wpdb;
+		$wpdb       = new Jobs_Idempotent_Insert_Test_Wpdb();
+		$wpdb->mode = $mode;
+		return array( new Jobs(), $wpdb );
+	};
+
+	list( $jobs, $wpdb ) = $new_jobs( 'success' );
+	$result              = $jobs->create_or_get_job(
+		array(
+			'label'           => str_repeat( 'x', 300 ),
+			'idempotency_key' => 'pipeline-batch:bounded-label',
+		)
+	);
+	$assert( is_array( $result ) && $result['created'], 'a valid idempotent job is created' );
+	$assert( 255 === mb_strlen( $wpdb->rows[1]['label'], 'UTF-8' ), 'label is bounded to its varchar(255) schema' );
+
+	list( $jobs, $wpdb ) = $new_jobs( 'duplicate_race' );
+	$result              = $jobs->create_or_get_job( array( 'idempotency_key' => 'pipeline-batch:race' ) );
+	$assert( is_array( $result ) && 77 === $result['job_id'], 'a delayed duplicate-race winner becomes canonical' );
+	$assert( 1 === $wpdb->insert_calls, 'duplicate errors do not retry the insert' );
+	$assert( 3 === $wpdb->lookup_calls, 'duplicate winner visibility polling is bounded' );
+
+	list( $jobs, $wpdb ) = $new_jobs( 'contention_then_success' );
+	$result              = $jobs->create_or_get_job( array( 'idempotency_key' => 'pipeline-batch:transient' ) );
+	$assert( is_array( $result ) && $result['created'], 'transient contention can recover' );
+	$assert( 3 === $wpdb->insert_calls, 'contention retries stop at the bounded successful attempt' );
+
+	$GLOBALS['job_insert_test_actions'] = array();
+	$GLOBALS['job_insert_test_cache']   = array();
+	list( $jobs, $wpdb )                = $new_jobs( 'contention' );
+	for ( $call = 0; $call < 5; ++$call ) {
+		$assert( false === $jobs->create_or_get_job( array( 'idempotency_key' => 'pipeline-batch:persistent' ) ), 'persistent contention remains visible to its caller' );
+	}
+	$logs    = $GLOBALS['job_insert_test_actions']['datamachine_log'] ?? array();
+	$signals = $GLOBALS['job_insert_test_actions']['datamachine_idempotent_job_insert_failed'] ?? array();
+	$assert( 15 === $wpdb->insert_calls, 'persistent contention is bounded to three attempts per call' );
+	$assert( 1 === count( $logs ), 'identical persistent failures produce one log per throttle window' );
+	$assert( 5 === count( $signals ), 'every persistent failure remains observable to metrics listeners' );
+	$assert( 3 === $logs[0][2]['attempts'], 'the final diagnostic records retry exhaustion' );
+	$assert( str_contains( $logs[0][2]['db_error'], 'Deadlock found' ), 'the original insert error survives recovery lookups' );
+	$assert( ! isset( $logs[0][2]['idempotency_key'] ), 'diagnostics do not expose raw idempotency keys' );
+
+	$GLOBALS['job_insert_test_actions'] = array();
+	$GLOBALS['job_insert_test_cache']   = array();
+	list( $jobs, $wpdb )                = $new_jobs( 'non_retryable' );
+	$assert( false === $jobs->create_or_get_job( array( 'idempotency_key' => 'pipeline-batch:invalid' ) ), 'non-retryable database failures return false' );
+	$assert( 1 === $wpdb->insert_calls, 'non-retryable database failures are not amplified' );
+	$log = $GLOBALS['job_insert_test_actions']['datamachine_log'][0][2] ?? array();
+	$assert( 'non_retryable' === ( $log['error_type'] ?? '' ), 'non-retryable errors are classified safely' );
+
+	echo "jobs-idempotent-insert-recovery-smoke: {$assertions} assertions passed.\n";
+}